htwkmooc/codeocean
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix usage of semaphore for replace_broken_container

Browse Source
This commit is contained in:
Sebastian Serth
2020-03-22 00:39:58 +01:00
parent 5be843ca00
commit 1c5bd91fec
1 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
lib/docker_container_pool.rb
View File

@ -4,6 +4,11 @@ require 'concurrent/timer_task'
class DockerContainerPool class DockerContainerPool
# Always use a semaphore when dealing with write access to `@containers` or `@all_containers` or read access if that involves creating a new container
# Some methods have the `bypass_semaphore` flag which should be rarely used. It was introduced when adding the semaphore to the `replace_broken_container` method.
# From there, it was added to `remove_from_all_containers` and `add_to_all_containers` first. That ensures, replacing is atomic.
# When `replace_broken_container` was extended to check the total amount of containers first, the `bypass_semaphore` flag was also added to `get_container` and
# back to `replace_broken_container`. When the semaphore is not released, no container operations can be done!
@semaphore = Concurrent::Semaphore.new(1) @semaphore = Concurrent::Semaphore.new(1)
@containers = Concurrent::Hash[ExecutionEnvironment.all.map { |execution_environment| [execution_environment.id, Concurrent::Array.new] }] @containers = Concurrent::Hash[ExecutionEnvironment.all.map { |execution_environment| [execution_environment.id, Concurrent::Array.new] }]
@ -74,12 +79,12 @@ class DockerContainerPool
@semaphore.release @semaphore.release
end end
def self.get_container(execution_environment) def self.get_container(execution_environment, bypass_semaphore: false)
# if pooling is active, do pooling, otherwise just create an container and return it # if pooling is active, do pooling, otherwise just create an container and return it
if config[:active] if config[:active]
@semaphore.acquire @semaphore.acquire unless bypass_semaphore
container = @containers[execution_environment.id].try(:shift) || nil container = @containers[execution_environment.id].try(:shift) || nil
@semaphore.release @semaphore.release unless bypass_semaphore
Rails.logger.debug('get_container fetched container ' + container.to_s + ' for execution environment ' + execution_environment.to_s) Rails.logger.debug('get_container fetched container ' + container.to_s + ' for execution environment ' + execution_environment.to_s)
# just access and the following if we got a container. Otherwise, the execution_environment might be just created and not fully exist yet. # just access and the following if we got a container. Otherwise, the execution_environment might be just created and not fully exist yet.
if(container) if(container)
@ -92,11 +97,11 @@ class DockerContainerPool
else else
Rails.logger.error('docker_container_pool.get_container retrieved a container not running. Container will be removed from list: ' + container.to_s) Rails.logger.error('docker_container_pool.get_container retrieved a container not running. Container will be removed from list: ' + container.to_s)
#TODO: check in which state the container actually is and treat it accordingly (dead,... => destroy?) #TODO: check in which state the container actually is and treat it accordingly (dead,... => destroy?)
container = replace_broken_container(container, execution_environment) container = replace_broken_container(container, execution_environment, bypass_semaphore: bypass_semaphore)
end end
rescue Docker::Error::NotFoundError => error rescue Docker::Error::NotFoundError => error
Rails.logger.error('docker_container_pool.get_container rescued from Docker::Error::NotFoundError. Most likely, the container is not there any longer. Removing faulty entry from list: ' + container.to_s) Rails.logger.error('docker_container_pool.get_container rescued from Docker::Error::NotFoundError. Most likely, the container is not there any longer. Removing faulty entry from list: ' + container.to_s)
container = replace_broken_container(container, execution_environment) container = replace_broken_container(container, execution_environment, bypass_semaphore: bypass_semaphore)
end end
end end
# returning nil is no problem. then the pool is just depleted. # returning nil is no problem. then the pool is just depleted.
@ -106,21 +111,21 @@ class DockerContainerPool
end end
end end
def self.replace_broken_container(container, execution_environment) def self.replace_broken_container(container, execution_environment, bypass_semaphore: false)
@semaphore.acquire # See note at the top for `bypass_semaphore`
@semaphore.acquire unless bypass_semaphore
remove_from_all_containers(container, execution_environment, bypass_semaphore: true) remove_from_all_containers(container, execution_environment, bypass_semaphore: true)
missing_counter_count = execution_environment.pool_size - @all_containers[execution_environment.id].length missing_counter_count = execution_environment.pool_size - @all_containers[execution_environment.id].length
if missing_counter_count > 0 if missing_counter_count > 0
Rails.logger.error('Creating a new container and returning that.') Rails.logger.error('Creating a new container and returning that.')
new_container = create_container(execution_environment) new_container = create_container(execution_environment)
DockerContainerPool.add_to_all_containers(new_container, execution_environment, bypass_semaphore: true) DockerContainerPool.add_to_all_containers(new_container, execution_environment, bypass_semaphore: true)
@semaphore.release
new_container
else else
Rails.logger.error('Broken container removed for ' + execution_environment.to_s + ' but not creating a new one. Currently, ' + missing_counter_count.abs + ' more containers than the configured pool size are available.') Rails.logger.error('Broken container removed for ' + execution_environment.to_s + ' but not creating a new one. Currently, ' + missing_counter_count.abs + ' more containers than the configured pool size are available.')
@semaphore.release new_container = get_container(execution_environment, bypass_semaphore: true)
get_container(execution_environment)
end end
@semaphore.release unless bypass_semaphore
new_container
end end
def self.quantities def self.quantities