@@ -2,6 +2,8 @@ * remove dependency on SyslogLogger so that starling works on windows. * fix config file loading so relative paths are expanded properly <jacob@jacobatzen.dk> * clean up redhat init.d script <gaffneyc@gmail.com> +* add a 'fetch' command that is non-blocking equivalent to 'get' <aaron.hawkins@shawkvalue.com> +* implement the 'delete' method to allow queues to be deleted <timshadel@gmail.com> == 0.9.8 * add fix to enable relative paths <david@motorator.com> CHANGELOG @@ -4,6 +4,7 @@ class Starling < MemCache WAIT_TIME = 0.25 alias_method :_original_get, :get + alias_method :_original_delete, :delete ## # fetch an item from a queue. @@ -24,6 +25,39 @@ class Starling < MemCache end ## + # Delete the key (queue) from all Starling servers. This is necessary + # because the random way a server is chosen in #get_server_for_key + # implies that the queue could easily be spread across the entire + # Starling cluster. + + def delete(key, expiry = 0) + with_servers do + _original_delete(key, expiry) + end + end + + ## + # Provides a way to work with a specific list of servers by + # forcing all calls to #get_server_for_key to use a specific + # server, and changing that server each time that the call + # yields to the block provided. This helps work around the + # normally random nature of the #get_server_for_key method. + # + # Acquires the mutex for the entire duration of the call + # since unrelated calls to #get_server_for_key might be + # adversely affected by the non_random result. + def with_servers(my_servers = @servers.dup) + return unless block_given? + with_lock do + my_servers.each do |server| + @force_server = server + yield + end + @force_server = nil + end + end + + ## # insert +value+ into +queue+. # # +expiry+ is expressed as a UNIX timestamp @@ -98,6 +132,7 @@ class Starling < MemCache raise ArgumentError, "illegal character in key #{key.inspect}" if key =~ /\s/ raise ArgumentError, "key too long #{key.inspect}" if key.length > 250 raise MemCacheError, "No servers available" if @servers.empty? + return @force_server if @force_server bukkits = @buckets.dup bukkits.nitems.times do |try| @@ -110,3 +145,24 @@ class Starling < MemCache raise MemCacheError, "No servers available (all dead)" end end + + +class MemCache + + protected + + ## + # Ensure that everything within the given block is executed + # within the locked mutex if this client is multithreaded. + # If the client isn't multithreaded, the block is simply executed. + def with_lock + return unless block_given? + begin + @mutex.lock if @multithread + yield + ensure + @mutex.unlock if @multithread + end + end + +end \ No newline at end of file lib/starling.rb @@ -22,6 +22,10 @@ module StarlingServer SET_RESPONSE_SUCCESS = "STORED\r\n".freeze SET_RESPONSE_FAILURE = "NOT STORED\r\n".freeze SET_CLIENT_DATA_ERROR = "CLIENT_ERROR bad data chunk\r\nERROR\r\n".freeze + + # DELETE Responses + DELETE_COMMAND = /\Adelete (.{1,250}) ([0-9]+)\r\n/m + DELETE_RESPONSE = "END\r\n".freeze # STAT Response STATS_COMMAND = /\Astats\r\n/m @@ -108,7 +112,6 @@ STAT queue_%s_age %d\r\n".freeze @data_buf = data return end - case data when SET_COMMAND @server.stats[:set_requests] += 1 @@ -121,6 +124,8 @@ STAT queue_%s_age %d\r\n".freeze when SHUTDOWN_COMMAND # no point in responding, they'll never get it. Runner::shutdown + when DELETE_COMMAND + delete $1 else logger.warn "Unknown command: #{data}." respond ERR_UNKNOWN_COMMAND @@ -136,6 +141,12 @@ STAT queue_%s_age %d\r\n".freeze end private + + def delete(queue) + @queue_collection.delete(queue) + respond DELETE_RESPONSE + end + def respond(str, *args) response = sprintf(str, *args) @server.stats[:bytes_written] += response.length lib/starling/handler.rb @@ -83,6 +83,11 @@ module StarlingServer @trx = nil @not_trx.close end + + def purge + close + File.delete(log_path) + end private lib/starling/persistent_queue.rb @@ -59,6 +59,12 @@ module StarlingServer @stats[:current_bytes] -= result.size result end + + def delete(key) + queue = @queues.delete(key) + return if queue.nil? + queue.purge + end ## # Returns all active queues. lib/starling/queue_collection.rb @@ -4,6 +4,7 @@ require 'rubygems' require 'fileutils' require 'memcache' require 'digest/md5' +require 'starling' require 'starling/server' @@ -51,6 +52,7 @@ describe "StarlingServer" do end @client = MemCache.new('127.0.0.1:22133') + end it "should test if temp_path exists and is writeable" do @@ -66,6 +68,15 @@ describe "StarlingServer" do @client.get('test_set_and_get_one_entry').should eql(v) end + it "should respond to delete" do + @client.delete("my_queue").should eql("END\r\n") + starling_client = Starling.new('127.0.0.1:22133') + starling_client.set('my_queue', 50) + starling_client.available_queues.size.should eql(1) + starling_client.delete("my_queue") + starling_client.available_queues.size.should eql(0) + end + it "should expire entries" do v = rand((2**32)-1) @client.get('test_set_with_expiry').should be_nil spec/starling_server_spec.rb 4b93086d9864e4ea9822cb67ba0461fac61bb0fb Tim Shadel github@timshadel.com http://github.com/defunkt/starling/commit/7fef96db63b925b80f6cf939536ac0d29305f427 7fef96db63b925b80f6cf939536ac0d29305f427 2008-10-10T06:27:01-07:00 2008-09-22T17:04:01-07:00 Implement the Delete call to delete a queue. * Client-side implementation ensures that the queue is deleted from every server in the cluster, since the Starling 'set' call will spread the items across all servers. Delegates to superclass for actual writing of delete request to socket, and manipulates #get_server_for_key to ensure that the queue is deleted from all servers. * When using the MemCache client, this isn't necessary since all queue items are put on the same Starling instance, so the 'delete' command is sent to that same server. * The server-side handler parses the standard MemCache delete message, so any MemCache client should be able to issue the command. * The rSpec test checks that the MemCache client can call the delete command, and also uses the Starling client to ensure that the number of available queues is appropriately updated. * Remove the queue's log file when deleting a queue. Otherwise the queue directory can get so full of log files (when frequently adding and removing queues) that the OS doesn't like it. Signed-off-by: Tim Shadel <github@timshadel.com> 7fc64c20b34b53c05084eb6db83480614f0d3246 Tim Shadel github@timshadel.com