alive?(pid)
click to toggle source
def alive? pid
pid = Integer("#{ pid }")
begin
Process::kill 0, pid
true
rescue Errno::ESRCH
false
end
end
attempt() { || ... }
click to toggle source
def attempt
ret = nil
loop{ break unless catch('attempt'){ ret = yield } == 'try_again' }
ret
end
create(path) { |f;| ... }
click to toggle source
def create path
umask = nil
f = nil
begin
umask = File::umask 022
f = open path, File::WRONLY|File::CREAT|File::EXCL, 0644
ensure
File::umask umask if umask
end
return(block_given? ? begin; yield f; ensure; f.close; end : f)
end
create_tmplock() { |f| ... }
click to toggle source
def create_tmplock
tmplock = tmpnam @dirname
begin
create(tmplock) do |f|
unless dont_use_lock_id
@lock_id = gen_lock_id
dumped = dump_lock_id
trace{"lock_id <\n#{ @lock_id.inspect }\n>"}
f.write dumped
f.flush
end
yield f
end
ensure
begin; File::unlink tmplock; rescue Errno::ENOENT; end if tmplock
end
end
dump_lock_id(lock_id = @lock_id)
click to toggle source
def dump_lock_id lock_id = @lock_id
"host: %s\npid: %s\nppid: %s\ntime: %s\n" %
lock_id.values_at('host','pid','ppid','time')
end
errmsg(e)
click to toggle source
def errmsg e
"%s (%s)\n%s\n" % [e.class, e.message, e.backtrace.join("\n")]
end
gen_lock_id()
click to toggle source
def gen_lock_id
Hash[
'host' => "#{ HOSTNAME }",
'pid' => "#{ Process.pid }",
'ppid' => "#{ Process.ppid }",
'time' => timestamp,
]
end
getopt(key, default = nil)
click to toggle source
def getopt key, default = nil
[ key, key.to_s, key.to_s.intern ].each do |k|
return @opts[k] if @opts.has_key?(k)
end
return default
end
give_up!()
click to toggle source
def give_up!
throw 'attempt', 'give_up'
end
load_lock_id(buf)
click to toggle source
def load_lock_id buf
lock_id = {}
kv = /([^:]+):(.*)/
buf.each do |line|
m = kv.match line
k, v = m[1], m[2]
next unless m and k and v
lock_id[k.strip] = v.strip
end
lock_id
end
lock() { |path| ... }
click to toggle source
def lock
raise StackingLockError, "<#{ @path }> is locked!" if @locked
sweep unless @dont_sweep
ret = nil
attempt do
begin
@sleep_cycle.reset
create_tmplock do |f|
begin
Timeout::timeout(@timeout) do
tmp_path = f.path
tmp_stat = f.lstat
n_retries = 0
trace{ "attempting to lock <#{ @path }>..." }
begin
i = 0
begin
trace{ "polling attempt <#{ i }>..." }
begin
File::link tmp_path, @path
rescue Errno::ENOENT
try_again!
end
lock_stat = File::lstat @path
raise StatLockError, "stat's do not agree" unless
tmp_stat.rdev == lock_stat.rdev and tmp_stat.ino == lock_stat.ino
trace{ "aquired lock <#{ @path }>" }
@locked = true
rescue => e
i += 1
unless i >= @poll_retries
t = [rand(@poll_max_sleep), @poll_max_sleep].min
trace{ "poll sleep <#{ t }>..." }
sleep t
retry
end
raise
end
rescue => e
n_retries += 1
trace{ "n_retries <#{ n_retries }>" }
case validlock?
when true
raise MaxTriesLockError, "surpased retries <#{ @retries }>" if
@retries and n_retries >= @retries
trace{ "found valid lock" }
sleeptime = @sleep_cycle.next
trace{ "sleep <#{ sleeptime }>..." }
sleep sleeptime
when false
trace{ "found invalid lock and removing" }
begin
File::unlink @path
@thief = true
warn "<#{ @path }> stolen by <#{ Process.pid }> at <#{ timestamp }>"
trace{ "i am a thief!" }
rescue Errno::ENOENT
end
trace{ "suspending <#{ @suspend }>" }
sleep @suspend
when nil
raise MaxTriesLockError, "surpased retries <#{ @retries }>" if
@retries and n_retries >= @retries
end
retry
end
end
rescue Timeout::Error
raise TimeoutLockError, "surpassed timeout <#{ @timeout }>"
end
end
if block_given?
stolen = false
refresher = (@refresh ? new_refresher : nil)
begin
begin
ret = yield @path
rescue StolenLockError
stolen = true
raise
end
ensure
begin
refresher.kill if refresher and refresher.status
ensure
unlock unless stolen
end
end
else
ObjectSpace.define_finalizer self, @clean if @clean
ret = self
end
rescue Errno::ESTALE, Errno::EIO => e
raise(NFSLockError, errmsg(e))
end
end
return ret
end
new_refresher()
click to toggle source
def new_refresher
Thread::new(Thread::current, @path, @refresh, @dont_use_lock_id) do |thread, path, refresh, dont_use_lock_id|
loop do
begin
touch path
trace{"touched <#{ path }> @ <#{ Time.now.to_f }>"}
unless dont_use_lock_id
loaded = load_lock_id(IO.read(path))
trace{"loaded <\n#{ loaded.inspect }\n>"}
raise unless loaded == @lock_id
end
sleep refresh
rescue Exception => e
trace{errmsg e}
thread.raise StolenLockError
Thread::exit
end
end
end
end
sweep()
click to toggle source
def sweep
begin
glob = File::join(@dirname, ".*lck")
paths = Dir[glob]
paths.each do |path|
begin
basename = File::basename path
pat = /^\s*\.([^_]+)_([^_]+)/
if pat.match(basename)
host, pid = $1, $2
else
next
end
host.gsub!(/^\.+|\.+$/,'')
quad = host.split /\./
host = quad.first
pat = /^\s*#{ host }/
if pat.match(HOSTNAME) and /^\s*\d+\s*$/.match(pid)
unless alive?(pid)
trace{ "process <#{ pid }> on <#{ host }> is no longer alive" }
trace{ "sweeping <#{ path }>" }
FileUtils::rm_f path
else
trace{ "process <#{ pid }> on <#{ host }> is still alive" }
trace{ "ignoring <#{ path }>" }
end
else
trace{ "ignoring <#{ path }> generated by <#{ host }>" }
end
rescue
next
end
end
rescue => e
warn(errmsg(e))
end
end
timestamp()
click to toggle source
def timestamp
time = Time.now
usec = time.usec.to_s
usec << '0' while usec.size < 6
"#{ time.strftime('%Y-%m-%d %H:%M:%S') }.#{ usec }"
end
tmpnam(dir, seed = File::basename($0))
click to toggle source
def tmpnam dir, seed = File::basename($0)
pid = Process.pid
time = Time.now
sec = time.to_i
usec = time.usec
"%s%s.%s_%d_%s_%d_%d_%d.lck" %
[dir, File::SEPARATOR, HOSTNAME, pid, seed, sec, usec, rand(sec)]
end
to_str()
click to toggle source
Also aliased as:
to_s, to_s
touch(path)
click to toggle source
def touch path
FileUtils.touch path
end
trace(s = nil) { || ... }
click to toggle source
def trace s = nil
STDERR.puts((s ? s : yield)) if @debug
end
try_again!()
click to toggle source
def try_again!
throw 'attempt', 'try_again'
end
Also aliased as:
again!, again!
uncache(file)
click to toggle source
def uncache file
refresh = nil
begin
is_a_file = File === file
path = (is_a_file ? file.path : file.to_s)
stat = (is_a_file ? file.stat : File::stat(file.to_s))
refresh = tmpnam(File::dirname(path))
File::link path, refresh
File::chmod stat.mode, path
File::utime stat.atime, stat.mtime, path
ensure
begin
File::unlink refresh if refresh
rescue Errno::ENOENT
end
end
end
unlock()
click to toggle source
def unlock
raise UnLockError, "<#{ @path }> is not locked!" unless @locked
begin
File::unlink @path
rescue Errno::ENOENT
raise StolenLockError, @path
ensure
@thief = false
@locked = false
ObjectSpace.undefine_finalizer self if @clean
end
end
validlock?()
click to toggle source
def validlock?
if @max_age
uncache @path rescue nil
begin
return((Time.now - File::stat(@path).mtime) < @max_age)
rescue Errno::ENOENT
return nil
end
else
exist = File::exist?(@path)
return(exist ? true : nil)
end
end
version()
click to toggle source
def version() VERSION end