58 lines
1.3 KiB
Ruby
58 lines
1.3 KiB
Ruby
require 'fileutils'
|
|
|
|
class DataPipeline
|
|
# another run holds the lock
|
|
class LockError < StandardError; end
|
|
|
|
#
|
|
# exclusive lock on a file, against overlapping runs (cron)
|
|
#
|
|
# lock 'data/.cudo.lock' -> in a pipeline: held from the start to the end of run
|
|
# etl run cudo.rb --lock=data/.lock -> for the whole script
|
|
#
|
|
# the lock is an flock on the file, released by the operating system when the process ends,
|
|
# also after a crash; the file stays and holds the pid of the last holder
|
|
#
|
|
class Lock
|
|
attr_reader :file
|
|
|
|
def initialize file
|
|
@file = file.to_s
|
|
end
|
|
|
|
def acquire
|
|
return self if @io
|
|
|
|
FileUtils.mkdir_p(File.dirname(@file))
|
|
io = File.open(@file, File::RDWR | File::CREAT, 0o644)
|
|
unless io.flock(File::LOCK_EX | File::LOCK_NB)
|
|
holder = io.read.strip
|
|
io.close
|
|
raise LockError, "#{@file} is locked#{" by pid #{holder}" unless holder.empty?}: another run is still active"
|
|
end
|
|
io.truncate(0)
|
|
io.write(Process.pid.to_s)
|
|
io.flush
|
|
@io = io
|
|
self
|
|
end
|
|
|
|
def release
|
|
return unless @io
|
|
|
|
@io.flock(File::LOCK_UN)
|
|
@io.close
|
|
@io = nil
|
|
end
|
|
|
|
def locked? = !@io.nil?
|
|
|
|
def synchronize
|
|
acquire
|
|
yield
|
|
ensure
|
|
release
|
|
end
|
|
end
|
|
end
|