Module: LockServer::ResourceLockServer

Defined in:
lib/lock-server.rb

Overview

Simple resource locking protocol implementation, suitable for EventMachine.

Constant Summary collapse

@@resource_queues =

Hash of resource queues. Keys are resource names, values are arrays of connections, trying to acquire resource.

{ }
@@mutexes =

Hash of mutexes used for syncronization of queue access. Keys are resource names, values are Mutex instances.

{ }

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#created_at ⇒ Object (readonly)

Returns the value of attribute created_at.



37
38
39
 
# File 'lib/lock-server.rb', line 37

def created_at
  @created_at
end

#peer_ip ⇒ Object (readonly)

Returns the value of attribute peer_ip.



36
37
38
 
# File 'lib/lock-server.rb', line 36

def peer_ip
  @peer_ip
end

#updated_at ⇒ Object (readonly)

Returns the value of attribute updated_at.



37
38
39
 
# File 'lib/lock-server.rb', line 37

def updated_at
  @updated_at
end

Instance Method Details

#allow_action ⇒ Object

Sends a notification that requested resource is free now.



136
137
138
 
# File 'lib/lock-server.rb', line 136

def allow_action
  send_line("ok #{@resource}")
end

#interrupt_timed_out_connection ⇒ Object 



113
114
115
 
# File 'lib/lock-server.rb', line 113

def interrupt_timed_out_connection
  return unless Time.now-@updated_at
end

#mutex_for(resource) ⇒ Object

Returns a mutex for specified resource

Parameters

resource

A name of resource which mutex should be returned



27
28
29
 
# File 'lib/lock-server.rb', line 27

def mutex_for(resource)
  @@mutexes[resource] ||= Mutex.new
end

#obtain_lock(resource) ⇒ Object

Creates a lock for requested resource, acquires mutex to protect resource queue manipulations and either notifies remote side about successful resource aquisition or tells it to wait. First connection in a queue is owning a resource, otherwise a connection will be put on hold for resource aquisition.

Parameters

resource

A name of resource to lock.



99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/lock-server.rb', line 99

def obtain_lock(resource)
  @resource = resource
  queue = queue_for(@resource)
  return if queue.include?(self)
  mutex_for(@resource).synchronize do
    queue << self
    if queue.size == 1
      allow_action
    else
      wait_for_lock_release
    end
  end
end

#post_init ⇒ Object 



39
40
41
42
43
 
# File 'lib/lock-server.rb', line 39

def post_init
  @resource = nil
  @peer_ip = Socket.unpack_sockaddr_in(self.get_peername)[1]
  set_comm_inactivity_timeout(LockServer::CONNECTION_TIMEOUT)
end

#queue_for(resource) ⇒ Object

Returns a queue for specified resource

Parameters

resource

A name of resource which query should be returned



17
18
19
20
 
# File 'lib/lock-server.rb', line 17

def queue_for(resource)
  p @@resource_queues
  @@resource_queues[resource] ||= []
end

#receive_data(data) ⇒ Object

A collback of EventMachine, being called when data is received. Used for processing of lock requests. To aquire lock for needed resource name application should send message 'lock resource_name', where resource_name is a name of needed resource. An answer will be either 'ok resource_name' (meaning lock have been obtained and application could proceed) or 'wait resource_name' (there's already lock on this resource, and application should wait). To unlock data, 'unlock' should be sent. This will remove current connection from a queue and a connection will be closed. Sending any other message will result in 'err: unknown message' and connection issued incorrect command will be closed.

Parameters

data

Data fetched from EventMachine



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/lock-server.rb', line 60

def receive_data(data)
  @updated_at = Time.now
  data.each_line do |line|
    command, resource = line.strip.split(' ')
    case command

    when 'lock'
      obtain_lock(resource)

    when 'unlock'
      send_line("released #{@resource}")
      close_connection_after_writing

    when "status"
      write_server_status
      close_connection_after_writing

    else
      send_line("err: unknown message")
      close_connection_after_writing
    end
  end
end

#release_lock ⇒ Object

Releases lock of previously specified resource. Calling this method will remove current connection from @resource aquisitors queue. Also, if current connection is owning a resource (is first in aquisitors queue), it will notify next connection that resource is free now.

Parameters

none



123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/lock-server.rb', line 123

def release_lock
  queue = queue_for(@resource)
  mutex_for(@resource).synchronize do
    if queue.first == self
      queue.shift
      queue.first.allow_action unless queue.empty?
    else
      queue.delete self
    end
  end
end

#send_line(string) ⇒ Object 



45
46
47
 
# File 'lib/lock-server.rb', line 45

def send_line(string)
  send_data(string+"\n")
end

#unbind ⇒ Object 



145
146
147
 
# File 'lib/lock-server.rb', line 145

def unbind
  release_lock
end

#wait_for_lock_release ⇒ Object

Sends a notification that requested resource is locked and an application should wait for lock.



141
142
143
 
# File 'lib/lock-server.rb', line 141

def wait_for_lock_release
  send_line("wait #{@resource}")
end

#write_server_status ⇒ Object 



84
85
86
87
88
89
90
 
# File 'lib/lock-server.rb', line 84

def write_server_status
  @@resource_queues.keys.each{ |k|
    next if queue_for(k).empty?
    send_line("Resource: #{k} (#{queue_for(k).size} connections waiting)")
    send_line(queue_for(k).collect(&:peer_ip).join(", "))
  }
end