Class: BucketFileCache

Inherits:
Object show all
Defined in:
lib/openc3/utilities/bucket_file_cache.rb

Constant Summary collapse

MAX_DISK_USAGE =

Default 20 GB

(ENV['OPENC3_BUCKET_FILE_CACHE_SIZE'] || 20_000_000_000).to_i
CHECK_TIME_SECONDS = 
3600
@@instance = 
nil
@@mutex = 
Mutex.new

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ BucketFileCache

Returns a new instance of BucketFileCache.



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 156

def initialize
  # Create local file cache location
  @cache_dir = Dir.mktmpdir
  FileUtils.mkdir_p(@cache_dir)
  at_exit do
    FileUtils.remove_dir(@cache_dir, true)
  end

  @current_disk_usage = 0
  @queued_bucket_files = []
  # Mirrors @queued_bucket_files as bucket_path => true. Membership is checked
  # once per file close in unreserve() and once per entry in age_out_files(),
  # both of which are O(cache size) linear scans against the bare Array.
  @queued_path_hash = {}
  @bucket_file_hash = {}
  bucket_file = nil

  # NOTE: check_time must be initialized outside the loop or it is pushed
  # forward on every iteration and the age out check below never fires
  check_time = Time.now + CHECK_TIME_SECONDS # Check for aged out files periodically

  @thread = Thread.new do
    client = OpenC3::Bucket.getClient()
    while true
      if @queued_bucket_files.length > 0 and @current_disk_usage < MAX_DISK_USAGE
        @@mutex.synchronize do
          bucket_file = @queued_bucket_files.shift
          @queued_path_hash.delete(bucket_file.bucket_path) if bucket_file
        end
        begin
          retrieved = bucket_file.retrieve(client)
          @@mutex.synchronize do
            @current_disk_usage += bucket_file.size if retrieved
          end
        rescue
          # Might have been deleted
        end
        sleep(0.01) # Small throttle
      else
        # Nothing to do or disk full
        if Time.now > check_time
          check_time = Time.now + CHECK_TIME_SECONDS
          age_out_files()
        end
        sleep(1)
      end
    end
  rescue => err
    OpenC3::Logger.error "BucketFileCache thread unexpectedly died\n#{err.formatted}"
  end
end

Instance Attribute Details

#cache_dir ⇒ Object (readonly)

Returns the value of attribute cache_dir.



143
144
145
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 143

def cache_dir
  @cache_dir
end

Class Method Details

.hint(bucket_paths) ⇒ Object 



208
209
210
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 208

def self.hint(bucket_paths)
  return instance().hint(bucket_paths)
end

.instance ⇒ Object 



148
149
150
151
152
153
154
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 148

def self.instance
  return @@instance if @@instance
  @@mutex.synchronize do
    @@instance ||= BucketFileCache.new
  end
  @@instance
end

.reserve(bucket_path) ⇒ Object 



212
213
214
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 212

def self.reserve(bucket_path)
  return instance().reserve(bucket_path)
end

.unreserve(bucket_file_or_path) ⇒ Object

Parameters:

  • BucketFile or its bucket path



217
218
219
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 217

def self.unreserve(bucket_file_or_path)
  return instance().unreserve(bucket_file_or_path)
end

Instance Method Details

#age_out_files ⇒ Object

Delete the local copy of any file that is old and no longer reserved. Files still waiting in the download queue are skipped because the download thread would re-add their size to @current_disk_usage after removal.



276
277
278
279
280
281
282
283
284
285
286
287
288
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 276

def age_out_files
  @@mutex.synchronize do
    @bucket_file_hash.delete_if do |bucket_path, bucket_file|
      next false if @queued_path_hash[bucket_path]
      if bucket_file.age_check
        @current_disk_usage -= bucket_file.size
        true
      else
        false
      end
    end
  end
end

#create_bucket_file(bucket_path) ⇒ Object 



290
291
292
293
294
295
296
297
298
299
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 290

def create_bucket_file(bucket_path)
  bucket_file = @bucket_file_hash[bucket_path]
  unless bucket_file
    bucket_file = BucketFile.new(bucket_path)
    @queued_bucket_files << bucket_file
    @queued_path_hash[bucket_path] = true
    @bucket_file_hash[bucket_path] = bucket_file
  end
  return bucket_file
end

#hint(bucket_paths) ⇒ Object 



221
222
223
224
225
226
227
228
229
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 221

def hint(bucket_paths)
  @@mutex.synchronize do
    bucket_paths.each_with_index do |bucket_path, index|
      bucket_file = create_bucket_file(bucket_path)
      bucket_file.priority = index
    end
    @queued_bucket_files.sort! {|file1, file2| file1.priority <=> file2.priority}
  end
end

#reserve(bucket_path) ⇒ Object 



231
232
233
234
235
236
237
238
239
240
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 231

def reserve(bucket_path)
  @@mutex.synchronize do
    bucket_file = create_bucket_file(bucket_path)
    retrieved = bucket_file.reserve
    @current_disk_usage += bucket_file.size if retrieved
    @queued_bucket_files.delete(bucket_file)
    @queued_path_hash.delete(bucket_path)
    return bucket_file
  end
end

#unreserve(bucket_file_or_path) ⇒ Object

Callers hold the BucketFile returned by reserve() but the cache is keyed by bucket path, so accept either and normalize to the path.

Parameters:

  • BucketFile or its bucket path



245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/openc3/utilities/bucket_file_cache.rb', line 245

def unreserve(bucket_file_or_path)
  if bucket_file_or_path.is_a?(BucketFile)
    released = bucket_file_or_path
    bucket_path = released.bucket_path
  else
    released = nil
    bucket_path = bucket_file_or_path
  end
  @@mutex.synchronize do
    bucket_file = @bucket_file_hash[bucket_path]
    return if bucket_file.nil?
    # Releasing to zero drops the entry, so the next reserve of the same path
    # builds a new BucketFile. A BucketFile that is no longer the entry for
    # its path was therefore already released, and releasing it again would
    # decrement its replacement and delete a local file another reader still
    # has open. Only the current owner of the path may release it.
    return unless released.nil? or released.equal?(bucket_file)

    bucket_file.unreserve
    if bucket_file.reservation_count <= 0 and !@queued_path_hash[bucket_path]
      @current_disk_usage -= bucket_file.size
      @bucket_file_hash.delete(bucket_path)
    end
  end
end