OpenSimulator GForge > Projects > Cable Beach Asset Server > SVN

Annotation of /trunk/lib/beitmemcached/MemcachedClient.cs

1 :	jhurliman	25	//Copyright (c) 2007-2008 Henrik Schr�der, Oliver Kofoed Pedersen
2 :
3 :			//Permission is hereby granted, free of charge, to any person
4 :			//obtaining a copy of this software and associated documentation
5 :			//files (the "Software"), to deal in the Software without
6 :			//restriction, including without limitation the rights to use,
7 :			//copy, modify, merge, publish, distribute, sublicense, and/or sell
8 :			//copies of the Software, and to permit persons to whom the
9 :			//Software is furnished to do so, subject to the following
10 :			//conditions:
11 :
12 :			//The above copyright notice and this permission notice shall be
13 :			//included in all copies or substantial portions of the Software.
14 :
15 :			//THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
16 :			//EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
17 :			//OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
18 :			//NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
19 :			//HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
20 :			//WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21 :			//FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
22 :			//OTHER DEALINGS IN THE SOFTWARE.
23 :
24 :			using System;
25 :			using System.Collections.Generic;
26 :			using System.Collections.Specialized;
27 :			using System.Configuration;
28 :			using System.Globalization;
29 :			using System.Text;
30 :
31 :			namespace BeIT.MemCached{
32 :			/// <summary>
33 :			/// Memcached client main class.
34 :			/// Use the static methods Setup and GetInstance to setup and get an instance of the client for use.
35 :			/// </summary>
36 :			public class MemcachedClient {
37 :			#region Static fields and methods.
38 :			private static Dictionary<string, MemcachedClient> instances = new Dictionary<string, MemcachedClient>();
39 :			private static LogAdapter logger = LogAdapter.GetLogger(typeof(MemcachedClient));
40 :
41 :			/// <summary>
42 :			/// Static method for creating an instance. This method will throw an exception if the name already exists.
43 :			/// </summary>
44 :			/// <param name="name">The name of the instance.</param>
45 :			/// <param name="servers">A list of memcached servers in standard notation: host:port.
46 :			/// If port is omitted, the default value of 11211 is used.
47 :			/// Both IP addresses and host names are accepted, for example:
48 :			/// "localhost", "127.0.0.1", "cache01.example.com:12345", "127.0.0.1:12345", etc.</param>
49 :			public static void Setup(string name, string[] servers) {
50 :			if (instances.ContainsKey(name)) {
51 :			throw new InvalidOperationException("Trying to configure MemcachedClient instance \"" + name + "\" twice.");
52 :			}
53 :			instances[name] = new MemcachedClient(name, servers);
54 :			}
55 :
56 :			/// <summary>
57 :			/// Static method which checks if a given named MemcachedClient instance exists.
58 :			/// </summary>
59 :			/// <param name="name">The name of the instance.</param>
60 :			/// <returns></returns>
61 :			public static bool Exists(string name) {
62 :			return instances.ContainsKey(name);
63 :			}
64 :
65 :			/// <summary>
66 :			/// Static method for getting the default instance named "default".
67 :			/// </summary>
68 :			private static MemcachedClient defaultInstance = null;
69 :			public static MemcachedClient GetInstance() {
70 :			return defaultInstance ?? (defaultInstance = GetInstance("default"));
71 :			}
72 :
73 :			/// <summary>
74 :			/// Static method for getting an instance.
75 :			/// This method will first check for a named instance that has been set up programmatically.
76 :			/// If that fails, an exception is thrown.
77 :			/// </summary>
78 :			/// <param name="name">The name of the instance.</param>
79 :			/// <returns>The named instance.</returns>
80 :			public static MemcachedClient GetInstance(string name) {
81 :			MemcachedClient c;
82 :			if (instances.TryGetValue(name, out c)) {
83 :			return c;
84 :			} else {
85 :			throw new InvalidOperationException("Unable to find MemcachedClient instance \"" + name + "\".");
86 :			}
87 :			}
88 :			#endregion
89 :
90 :			#region Fields, constructors, and private methods.
91 :			public readonly string Name;
92 :			private readonly ServerPool serverPool;
93 :
94 :			/// <summary>
95 :			/// If you specify a key prefix, it will be appended to all keys before they are sent to the memcached server.
96 :			/// They key prefix is not used when calculating which server a key belongs to.
97 :			/// </summary>
98 :			public string KeyPrefix { get { return keyPrefix; } set { keyPrefix = value; } }
99 :			private string keyPrefix = "";
100 :
101 :			/// <summary>
102 :			/// The send receive timeout is used to determine how long the client should wait for data to be sent
103 :			/// and received from the server, specified in milliseconds. The default value is 2000.
104 :			/// </summary>
105 :			public int SendReceiveTimeout { get { return serverPool.SendReceiveTimeout; } set { serverPool.SendReceiveTimeout = value; } }
106 :
107 :			/// <summary>
108 :			/// The min pool size determines the number of sockets the socket pool will keep.
109 :			/// Note that no sockets will be created on startup, only on use, so the socket pool will only
110 :			/// contain this amount of sockets if the amount of simultaneous requests goes above it.
111 :			/// The default value is 5.
112 :			/// </summary>
113 :			public uint MinPoolSize {
114 :			get { return serverPool.MinPoolSize; }
115 :			set {
116 :			if (value > MaxPoolSize) { throw new ArgumentException("MinPoolSize (" + value + ") may not be larger than the MaxPoolSize (" + MaxPoolSize + ")."); }
117 :			serverPool.MinPoolSize = value;
118 :			}
119 :			}
120 :
121 :			/// <summary>
122 :			/// The max pool size determines how large the socket connection pool is allowed to grow.
123 :			/// There can be more sockets in use than this amount, but when the extra sockets are returned, they will be destroyed.
124 :			/// The default value is 10.
125 :			/// </summary>
126 :			public uint MaxPoolSize {
127 :			get { return serverPool.MaxPoolSize; }
128 :			set {
129 :			if (value < MinPoolSize) { throw new ArgumentException("MaxPoolSize (" + value + ") may not be smaller than the MinPoolSize (" + MinPoolSize + ")."); }
130 :			serverPool.MaxPoolSize = value;
131 :			}
132 :			}
133 :
134 :			/// <summary>
135 :			/// If the pool contains more than the minimum amount of sockets, and a socket is returned that is older than this recycle age
136 :			/// that socket will be destroyed instead of put back in the pool. This allows the pool to shrink back to the min pool size after a peak in usage.
137 :			/// The default value is 30 minutes.
138 :			/// </summary>
139 :			public TimeSpan SocketRecycleAge { get { return serverPool.SocketRecycleAge; } set { serverPool.SocketRecycleAge = value; } }
140 :
141 :			private uint compressionThreshold = 1024*128; //128kb
142 :			/// <summary>
143 :			/// If an object being stored is larger in bytes than the compression threshold, it will internally be compressed before begin stored,
144 :			/// and it will transparently be decompressed when retrieved. Only strings, byte arrays and objects can be compressed.
145 :			/// The default value is 1048576 bytes = 1MB.
146 :			/// </summary>
147 :			public uint CompressionThreshold { get { return compressionThreshold; } set { compressionThreshold = value; } }
148 :
149 :			//Private constructor
150 :			private MemcachedClient(string name, string[] hosts) {
151 :			if (String.IsNullOrEmpty(name)) {
152 :			throw new ArgumentException("Name of MemcachedClient instance cannot be empty.");
153 :			}
154 :			if (hosts == null \|\| hosts.Length == 0) {
155 :			throw new ArgumentException("Cannot configure MemcachedClient with empty list of hosts.");
156 :			}
157 :
158 :			Name = name;
159 :			serverPool = new ServerPool(hosts);
160 :			}
161 :
162 :			/// <summary>
163 :			/// Private key hashing method that uses the modified FNV hash.
164 :			/// </summary>
165 :			/// <param name="key">The key to hash.</param>
166 :			/// <returns>The hashed key.</returns>
167 :			private uint hash(string key) {
168 :			checkKey(key);
169 :			return BitConverter.ToUInt32(new ModifiedFNV1_32().ComputeHash(Encoding.UTF8.GetBytes(key)), 0);
170 :			}
171 :
172 :			/// <summary>
173 :			/// Private hashing method for user-supplied hash values.
174 :			/// </summary>
175 :			/// <param name="hashvalue">The user-supplied hash value to hash.</param>
176 :			/// <returns>The hashed value</returns>
177 :			private uint hash(uint hashvalue) {
178 :			return BitConverter.ToUInt32(new ModifiedFNV1_32().ComputeHash(BitConverter.GetBytes(hashvalue)), 0);
179 :			}
180 :
181 :			/// <summary>
182 :			/// Private multi-hashing method.
183 :			/// </summary>
184 :			/// <param name="keys">An array of keys to hash.</param>
185 :			/// <returns>An arrays of hashes.</returns>
186 :			private uint[] hash(string[] keys) {
187 :			uint[] result = new uint[keys.Length];
188 :			for (int i = 0; i < keys.Length; i++) {
189 :			result[i] = hash(keys[i]);
190 :			}
191 :			return result;
192 :			}
193 :
194 :			/// <summary>
195 :			/// Private multi-hashing method for user-supplied hash values.
196 :			/// </summary>
197 :			/// <param name="hashvalues">An array of keys to hash.</param>
198 :			/// <returns>An arrays of hashes.</returns>
199 :			private uint[] hash(uint[] hashvalues) {
200 :			uint[] result = new uint[hashvalues.Length];
201 :			for (int i = 0; i < hashvalues.Length; i++) {
202 :			result[i] = hash(hashvalues[i]);
203 :			}
204 :			return result;
205 :			}
206 :
207 :			/// <summary>
208 :			/// Private key-checking method.
209 :			/// Throws an exception if the key does not conform to memcached protocol requirements:
210 :			/// It may not contain whitespace, it may not be null or empty, and it may not be longer than 250 characters.
211 :			/// </summary>
212 :			/// <param name="key">The key to check.</param>
213 :			private void checkKey(string key) {
214 :			if (key == null) {
215 :			throw new ArgumentNullException("Key may not be null.");
216 :			}
217 :			if (key.Length == 0) {
218 :			throw new ArgumentException("Key may not be empty.");
219 :			}
220 :			if (key.Length > 250) {
221 :			throw new ArgumentException("Key may not be longer than 250 characters.");
222 :			}
223 :			if (key.Contains(" ") \|\| key.Contains("\n") \|\| key.Contains("\r") \|\| key.Contains("\t") \|\| key.Contains("\f") \|\| key.Contains("\v")) {
224 :			throw new ArgumentException("Key may not contain whitespace or control characters.");
225 :			}
226 :			}
227 :
228 :			//Private Unix-time converter
229 :			private static DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
230 :			private static int getUnixTime(DateTime datetime) {
231 :			return (int)(datetime.ToUniversalTime() - epoch).TotalSeconds;
232 :			}
233 :			#endregion
234 :
235 :			#region Set, Add, and Replace.
236 :			/// <summary>
237 :			/// This method corresponds to the "set" command in the memcached protocol.
238 :			/// It will unconditionally set the given key to the given value.
239 :			/// Using the overloads it is possible to specify an expiry time, either relative as a TimeSpan or
240 :			/// absolute as a DateTime. It is also possible to specify a custom hash to override server selection.
241 :			/// This method returns true if the value was successfully set.
242 :			/// </summary>
243 :			public bool Set(string key, object value) { return store("set", key, true, value, hash(key), 0); }
244 :			public bool Set(string key, object value, uint hash) { return store("set", key, false, value, this.hash(hash), 0); }
245 :			public bool Set(string key, object value, TimeSpan expiry) { return store("set", key, true, value, hash(key), (int)expiry.TotalSeconds); }
246 :			public bool Set(string key, object value, uint hash, TimeSpan expiry) { return store("set", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
247 :			public bool Set(string key, object value, DateTime expiry) { return store("set", key, true, value, hash(key), getUnixTime(expiry)); }
248 :			public bool Set(string key, object value, uint hash, DateTime expiry) { return store("set", key, false, value, this.hash(hash), getUnixTime(expiry)); }
249 :
250 :			/// <summary>
251 :			/// This method corresponds to the "add" command in the memcached protocol.
252 :			/// It will set the given key to the given value only if the key does not already exist.
253 :			/// Using the overloads it is possible to specify an expiry time, either relative as a TimeSpan or
254 :			/// absolute as a DateTime. It is also possible to specify a custom hash to override server selection.
255 :			/// This method returns true if the value was successfully added.
256 :			/// </summary>
257 :			public bool Add(string key, object value) { return store("add", key, true, value, hash(key), 0); }
258 :			public bool Add(string key, object value, uint hash) { return store("add", key, false, value, this.hash(hash), 0); }
259 :			public bool Add(string key, object value, TimeSpan expiry) { return store("add", key, true, value, hash(key), (int)expiry.TotalSeconds); }
260 :			public bool Add(string key, object value, uint hash, TimeSpan expiry) { return store("add", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
261 :			public bool Add(string key, object value, DateTime expiry) { return store("add", key, true, value, hash(key), getUnixTime(expiry)); }
262 :			public bool Add(string key, object value, uint hash, DateTime expiry) { return store("add", key, false, value, this.hash(hash), getUnixTime(expiry)); }
263 :
264 :			/// <summary>
265 :			/// This method corresponds to the "replace" command in the memcached protocol.
266 :			/// It will set the given key to the given value only if the key already exists.
267 :			/// Using the overloads it is possible to specify an expiry time, either relative as a TimeSpan or
268 :			/// absolute as a DateTime. It is also possible to specify a custom hash to override server selection.
269 :			/// This method returns true if the value was successfully replaced.
270 :			/// </summary>
271 :			public bool Replace(string key, object value) { return store("replace", key, true, value, hash(key), 0); }
272 :			public bool Replace(string key, object value, uint hash) { return store("replace", key, false, value, this.hash(hash), 0); }
273 :			public bool Replace(string key, object value, TimeSpan expiry) { return store("replace", key, true, value, hash(key), (int)expiry.TotalSeconds); }
274 :			public bool Replace(string key, object value, uint hash, TimeSpan expiry) { return store("replace", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
275 :			public bool Replace(string key, object value, DateTime expiry) { return store("replace", key, true, value, hash(key), getUnixTime(expiry)); }
276 :			public bool Replace(string key, object value, uint hash, DateTime expiry) { return store("replace", key, false, value, this.hash(hash), getUnixTime(expiry)); }
277 :
278 :			/// <summary>
279 :			/// This method corresponds to the "append" command in the memcached protocol.
280 :			/// It will append the given value to the given key, if the key already exists.
281 :			/// Modifying a key with this command will not change its expiry time.
282 :			/// Using the overload it is possible to specify a custom hash to override server selection.
283 :			/// </summary>
284 :			public bool Append(string key, object value) { return store("append", key, true, value, hash(key)); }
285 :			public bool Append(string key, object value, uint hash) { return store("append", key, false, value, this.hash(hash)); }
286 :
287 :			/// <summary>
288 :			/// This method corresponds to the "prepend" command in the memcached protocol.
289 :			/// It will prepend the given value to the given key, if the key already exists.
290 :			/// Modifying a key with this command will not change its expiry time.
291 :			/// Using the overload it is possible to specify a custom hash to override server selection.
292 :			/// </summary>
293 :			public bool Prepend(string key, object value) { return store("prepend", key, true, value, hash(key)); }
294 :			public bool Prepend(string key, object value, uint hash) { return store("prepend", key, false, value, this.hash(hash)); }
295 :
296 :			public enum CasResult {
297 :			Stored = 0,
298 :			NotStored = 1,
299 :			Exists = 2,
300 :			NotFound = 3
301 :			}
302 :
303 :			public CasResult CheckAndSet(string key, object value, ulong unique) { return store(key, true, value, hash(key), 0, unique); }
304 :			public CasResult CheckAndSet(string key, object value, uint hash, ulong unique) { return store(key, false, value, this.hash(hash), 0, unique); }
305 :			public CasResult CheckAndSet(string key, object value, TimeSpan expiry, ulong unique) { return store(key, true, value, hash(key), (int)expiry.TotalSeconds, unique); }
306 :			public CasResult CheckAndSet(string key, object value, uint hash, TimeSpan expiry, ulong unique) { return store(key, false, value, this.hash(hash), (int)expiry.TotalSeconds, unique); }
307 :			public CasResult CheckAndSet(string key, object value, DateTime expiry, ulong unique) { return store(key, true, value, hash(key), getUnixTime(expiry), unique); }
308 :			public CasResult CheckAndSet(string key, object value, uint hash, DateTime expiry, ulong unique) { return store(key, false, value, this.hash(hash), getUnixTime(expiry), unique); }
309 :
310 :			//Private overload for the Set, Add and Replace commands.
311 :			private bool store(string command, string key, bool keyIsChecked, object value, uint hash, int expiry) {
312 :			return store(command, key, keyIsChecked, value, hash, expiry, 0).StartsWith("STORED");
313 :			}
314 :
315 :			//Private overload for the Append and Prepend commands.
316 :			private bool store(string command, string key, bool keyIsChecked, object value, uint hash) {
317 :			return store(command, key, keyIsChecked, value, hash, 0, 0).StartsWith("STORED");
318 :			}
319 :
320 :			//Private overload for the Cas command.
321 :			private CasResult store(string key, bool keyIsChecked, object value, uint hash, int expiry, ulong unique) {
322 :			string result = store("cas", key, keyIsChecked, value, hash, expiry, unique);
323 :			if (result.StartsWith("STORED")) {
324 :			return CasResult.Stored;
325 :			} else if (result.StartsWith("EXISTS")) {
326 :			return CasResult.Exists;
327 :			} else if (result.StartsWith("NOT_FOUND")) {
328 :			return CasResult.NotFound;
329 :			}
330 :			return CasResult.NotStored;
331 :			}
332 :
333 :			//Private common store method.
334 :			private string store(string command, string key, bool keyIsChecked, object value, uint hash, int expiry, ulong unique) {
335 :			if (!keyIsChecked) {
336 :			checkKey(key);
337 :			}
338 :
339 :			return serverPool.Execute<string>(hash, "", delegate(PooledSocket socket) {
340 :			SerializedType type;
341 :			byte[] bytes;
342 :
343 :			//Serialize object efficiently, store the datatype marker in the flags property.
344 :			try {
345 :			bytes = Serializer.Serialize(value, out type, CompressionThreshold);
346 :			} catch (Exception e) {
347 :			//If serialization fails, return false;
348 :
349 :			logger.Error("Error serializing object for key '" + key + "'.", e);
350 :			return "";
351 :			}
352 :
353 :			//Create commandline
354 :			string commandline = "";
355 :			switch(command) {
356 :			case "set":
357 :			case "add":
358 :			case "replace":
359 :			commandline = command + " " + keyPrefix + key + " " + (ushort)type + " " + expiry + " " + bytes.Length + "\r\n";
360 :			break;
361 :			case "append":
362 :			case "prepend":
363 :			commandline = command + " " + keyPrefix + key + " 0 0 " + bytes.Length + "\r\n";
364 :			break;
365 :			case "cas":
366 :			commandline = command + " " + keyPrefix + key + " " + (ushort)type + " " + expiry + " " + bytes.Length + " " + unique + "\r\n";
367 :			break;
368 :			}
369 :
370 :			//Write commandline and serialized object.
371 :			socket.Write(commandline);
372 :			socket.Write(bytes);
373 :			socket.Write("\r\n");
374 :			return socket.ReadResponse();
375 :			});
376 :			}
377 :
378 :			#endregion
379 :
380 :			#region Get
381 :			/// <summary>
382 :			/// This method corresponds to the "get" command in the memcached protocol.
383 :			/// It will return the value for the given key. It will return null if the key did not exist,
384 :			/// or if it was unable to retrieve the value.
385 :			/// If given an array of keys, it will return a same-sized array of objects with the corresponding
386 :			/// values.
387 :			/// Use the overload to specify a custom hash to override server selection.
388 :			/// </summary>
389 :			public object Get(string key) { ulong i; return get("get", key, true, hash(key), out i); }
390 :			public object Get(string key, uint hash) { ulong i; return get("get", key, false, this.hash(hash), out i); }
391 :
392 :			/// <summary>
393 :			/// This method corresponds to the "gets" command in the memcached protocol.
394 :			/// It works exactly like the Get method, but it will also return the cas unique value for the item.
395 :			/// </summary>
396 :			public object Gets(string key, out ulong unique) { return get("gets", key, true, hash(key), out unique); }
397 :			public object Gets(string key, uint hash, out ulong unique) { return get("gets", key, false, this.hash(hash), out unique); }
398 :
399 :			private object get(string command, string key, bool keyIsChecked, uint hash, out ulong unique) {
400 :			if (!keyIsChecked) {
401 :			checkKey(key);
402 :			}
403 :
404 :			ulong __unique = 0;
405 :			object value = serverPool.Execute<object>(hash, null, delegate(PooledSocket socket) {
406 :			socket.Write(command + " " + keyPrefix + key + "\r\n");
407 :			object _value;
408 :			ulong _unique;
409 :			if (readValue(socket, out _value, out key, out _unique)) {
410 :			socket.ReadLine(); //Read the trailing END.
411 :			}
412 :			__unique = _unique;
413 :			return _value;
414 :			});
415 :			unique = __unique;
416 :			return value;
417 :			}
418 :
419 :			/// <summary>
420 :			/// This method executes a multi-get. It will group the keys by server and execute a single get
421 :			/// for each server, and combine the results. The returned object[] will have the same size as
422 :			/// the given key array, and contain either null or a value at each position according to
423 :			/// the key on that position.
424 :			/// </summary>
425 :			public object[] Get(string[] keys) { ulong[] uniques; return get("get", keys, true, hash(keys), out uniques); }
426 :			public object[] Get(string[] keys, uint[] hashes) { ulong[] uniques; return get("get", keys, false, hash(hashes), out uniques); }
427 :
428 :			/// <summary>
429 :			/// This method does a multi-gets. It functions exactly like the multi-get method, but it will
430 :			/// also return an array of cas unique values as an out parameter.
431 :			/// </summary>
432 :			public object[] Gets(string[] keys, out ulong[] uniques) { return get("gets", keys, true, hash(keys), out uniques); }
433 :			public object[] Gets(string[] keys, uint[] hashes, out ulong[] uniques) { return get("gets", keys, false, hash(hashes), out uniques); }
434 :
435 :			private object[] get(string command, string[] keys, bool keysAreChecked, uint[] hashes, out ulong[] uniques) {
436 :			//Check arguments.
437 :			if (keys == null \|\| hashes == null) {
438 :			throw new ArgumentException("Keys and hashes arrays must not be null.");
439 :			}
440 :			if (keys.Length != hashes.Length) {
441 :			throw new ArgumentException("Keys and hashes arrays must be of the same length.");
442 :			}
443 :			uniques = new ulong[keys.Length];
444 :
445 :			//Avoid going through the server grouping if there's only one key.
446 :			if (keys.Length == 1) {
447 :			return new object[] { get(command, keys[0], keysAreChecked, hashes[0], out uniques[0]) };
448 :			}
449 :
450 :			//Check keys.
451 :			if (!keysAreChecked) {
452 :			for (int i = 0; i < keys.Length; i++) {
453 :			checkKey(keys[i]);
454 :			}
455 :			}
456 :
457 :			//Group the keys/hashes by server(pool)
458 :			Dictionary<SocketPool, Dictionary<string, List<int>>> dict = new Dictionary<SocketPool, Dictionary<string, List<int>>>();
459 :			for (int i = 0; i < keys.Length; i++) {
460 :			Dictionary<string, List<int>> getsForServer;
461 :			SocketPool pool = serverPool.GetSocketPool(hashes[i]);
462 :			if (!dict.TryGetValue(pool, out getsForServer)) {
463 :			dict[pool] = getsForServer = new Dictionary<string, List<int>>();
464 :			}
465 :
466 :			List<int> positions;
467 :			if(!getsForServer.TryGetValue(keys[i], out positions)){
468 :			getsForServer[keys[i]] = positions = new List<int>();
469 :			}
470 :			positions.Add(i);
471 :			}
472 :
473 :			//Get the values
474 :			object[] returnValues = new object[keys.Length];
475 :			ulong[] _uniques = new ulong[keys.Length];
476 :			foreach (KeyValuePair<SocketPool, Dictionary<string, List<int>>> kv in dict) {
477 :			serverPool.Execute(kv.Key, delegate(PooledSocket socket){
478 :			//Build the get request
479 :			StringBuilder getRequest = new StringBuilder(command);
480 :			foreach (KeyValuePair<string, List<int>> key in kv.Value) {
481 :			getRequest.Append(" ");
482 :			getRequest.Append(keyPrefix);
483 :			getRequest.Append(key.Key);
484 :			}
485 :			getRequest.Append("\r\n");
486 :
487 :			//Send get request
488 :			socket.Write(getRequest.ToString());
489 :
490 :			//Read values, one by one
491 :			object gottenObject;
492 :			string gottenKey;
493 :			ulong unique;
494 :			while (readValue(socket, out gottenObject, out gottenKey, out unique)) {
495 :			foreach(int position in kv.Value[gottenKey]) {
496 :			returnValues[position] = gottenObject;
497 :			_uniques[position] = unique;
498 :			}
499 :			}
500 :			});
501 :			}
502 :			uniques = _uniques;
503 :			return returnValues;
504 :			}
505 :
506 :			//Private method for reading results of the "get" command.
507 :			private bool readValue(PooledSocket socket, out object value, out string key, out ulong unique) {
508 :			string response = socket.ReadResponse();
509 :			string[] parts = response.Split(' '); //Result line from server: "VALUE <key> <flags> <bytes> <cas unique>"
510 :			if (parts[0] == "VALUE") {
511 :			key = parts[1];
512 :			SerializedType type = (SerializedType)Enum.Parse(typeof(SerializedType), parts[2]);
513 :			byte[] bytes = new byte[Convert.ToUInt32(parts[3], CultureInfo.InvariantCulture)];
514 :			if (parts.Length > 4) {
515 :			unique = Convert.ToUInt64(parts[4]);
516 :			} else {
517 :			unique = 0;
518 :			}
519 :			socket.Read(bytes);
520 :			socket.SkipUntilEndOfLine(); //Skip the trailing \r\n
521 :			try {
522 :			value = Serializer.DeSerialize(bytes, type);
523 :			} catch (Exception e) {
524 :			//If deserialization fails, return null
525 :			value = null;
526 :			logger.Error("Error deserializing object for key '" + key + "' of type " + type + ".", e);
527 :			}
528 :			return true;
529 :			} else {
530 :			key = null;
531 :			value = null;
532 :			unique = 0;
533 :			return false;
534 :			}
535 :			}
536 :			#endregion
537 :
538 :			#region Delete
539 :			/// <summary>
540 :			/// This method corresponds to the "delete" command in the memcache protocol.
541 :			/// It will immediately delete the given key and corresponding value.
542 :			/// Use the overloads to specify an amount of time the item should be in the delete queue on the server,
543 :			/// or to specify a custom hash to override server selection.
544 :			/// </summary>
545 :			public bool Delete(string key) { return delete(key, true, hash(key), 0); }
546 :			public bool Delete(string key, uint hash) { return delete(key, false, this.hash(hash), 0); }
547 :			public bool Delete(string key, TimeSpan delay) { return delete(key, true, hash(key), (int)delay.TotalSeconds); }
548 :			public bool Delete(string key, uint hash, TimeSpan delay) { return delete(key, false, this.hash(hash), (int)delay.TotalSeconds); }
549 :			public bool Delete(string key, DateTime delay) { return delete(key, true, hash(key), getUnixTime(delay)); }
550 :			public bool Delete(string key, uint hash, DateTime delay) { return delete(key, false, this.hash(hash), getUnixTime(delay)); }
551 :
552 :			private bool delete(string key, bool keyIsChecked, uint hash, int time) {
553 :			if (!keyIsChecked) {
554 :			checkKey(key);
555 :			}
556 :
557 :			return serverPool.Execute<bool>(hash, false, delegate(PooledSocket socket){
558 :			string commandline;
559 :			if (time == 0) {
560 :			commandline = "delete " + keyPrefix + key + "\r\n";
561 :			} else {
562 :			commandline = "delete " + keyPrefix + key + " " + time + "\r\n";
563 :			}
564 :			socket.Write(commandline);
565 :			return socket.ReadResponse().StartsWith("DELETED");
566 :			});
567 :			}
568 :			#endregion
569 :
570 :			#region Increment Decrement
571 :			/// <summary>
572 :			/// This method sets the key to the given value, and stores it in a format such that the methods
573 :			/// Increment and Decrement can be used successfully on it, i.e. decimal representation of a 64-bit unsigned integer.
574 :			/// Using the overloads it is possible to specify an expiry time, either relative as a TimeSpan or
575 :			/// absolute as a DateTime. It is also possible to specify a custom hash to override server selection.
576 :			/// This method returns true if the counter was successfully set.
577 :			/// </summary>
578 :			public bool SetCounter(string key, ulong value) { return Set(key, value.ToString(CultureInfo.InvariantCulture)); }
579 :			public bool SetCounter(string key, ulong value, uint hash) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash)); }
580 :			public bool SetCounter(string key, ulong value, TimeSpan expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }
581 :			public bool SetCounter(string key, ulong value, uint hash, TimeSpan expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash), expiry); }
582 :			public bool SetCounter(string key, ulong value, DateTime expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }
583 :			public bool SetCounter(string key, ulong value, uint hash, DateTime expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash), expiry); }
584 :
585 :			/// <summary>
586 :			/// This method returns the value for the given key as a ulong?, a nullable 64-bit unsigned integer.
587 :			/// It returns null if the item did not exist, was not stored properly as per the SetCounter method, or
588 :			/// if it was not able to successfully retrieve the item.
589 :			/// </summary>
590 :			public ulong? GetCounter(string key) {return getCounter(key, true, hash(key));}
591 :			public ulong? GetCounter(string key, uint hash) { return getCounter(key, false, this.hash(hash)); }
592 :
593 :			private ulong? getCounter(string key, bool keyIsChecked, uint hash) {
594 :			ulong parsedLong, unique;
595 :			return ulong.TryParse(get("get", key, keyIsChecked, hash, out unique) as string, out parsedLong) ? (ulong?)parsedLong : null;
596 :			}
597 :
598 :			public ulong?[] GetCounter(string[] keys) {return getCounter(keys, true, hash(keys));}
599 :			public ulong?[] GetCounter(string[] keys, uint[] hashes) { return getCounter(keys, false, hash(hashes)); }
600 :
601 :			private ulong?[] getCounter(string[] keys, bool keysAreChecked, uint[] hashes) {
602 :			ulong?[] results = new ulong?[keys.Length];
603 :			ulong[] uniques;
604 :			object[] values = get("get", keys, keysAreChecked, hashes, out uniques);
605 :			for (int i = 0; i < values.Length; i++) {
606 :			ulong parsedLong;
607 :			results[i] = ulong.TryParse(values[i] as string, out parsedLong) ? (ulong?)parsedLong : null;
608 :			}
609 :			return results;
610 :			}
611 :
612 :			/// <summary>
613 :			/// This method corresponds to the "incr" command in the memcached protocol.
614 :			/// It will increase the item with the given value and return the new value.
615 :			/// It will return null if the item did not exist, was not stored properly as per the SetCounter method, or
616 :			/// if it was not able to successfully retrieve the item.
617 :			/// </summary>
618 :			public ulong? Increment(string key, ulong value) { return incrementDecrement("incr", key, true, value, hash(key)); }
619 :			public ulong? Increment(string key, ulong value, uint hash) { return incrementDecrement("incr", key, false, value, this.hash(hash)); }
620 :
621 :			/// <summary>
622 :			/// This method corresponds to the "decr" command in the memcached protocol.
623 :			/// It will decrease the item with the given value and return the new value. If the new value would be
624 :			/// less than 0, it will be set to 0, and the method will return 0.
625 :			/// It will return null if the item did not exist, was not stored properly as per the SetCounter method, or
626 :			/// if it was not able to successfully retrieve the item.
627 :			/// </summary>
628 :			public ulong? Decrement(string key, ulong value) { return incrementDecrement("decr", key, true, value, hash(key)); }
629 :			public ulong? Decrement(string key, ulong value, uint hash) { return incrementDecrement("decr", key, false, value, this.hash(hash)); }
630 :
631 :			private ulong? incrementDecrement(string cmd, string key, bool keyIsChecked, ulong value, uint hash) {
632 :			if (!keyIsChecked) {
633 :			checkKey(key);
634 :			}
635 :			return serverPool.Execute<ulong?>(hash, null, delegate(PooledSocket socket) {
636 :			string command = cmd + " " + keyPrefix + key + " " + value + "\r\n";
637 :			socket.Write(command);
638 :			string response = socket.ReadResponse();
639 :			if (response.StartsWith("NOT_FOUND")) {
640 :			return null;
641 :			} else {
642 :			return Convert.ToUInt64(response.TrimEnd('\0', '\r', '\n'));
643 :			}
644 :			});
645 :			}
646 :			#endregion
647 :
648 :			#region Flush All
649 :			/// <summary>
650 :			/// This method corresponds to the "flush_all" command in the memcached protocol.
651 :			/// When this method is called, it will send the flush command to all servers, thereby deleting
652 :			/// all items on all servers.
653 :			/// Use the overloads to set a delay for the flushing. If the parameter staggered is set to true,
654 :			/// the client will increase the delay for each server, i.e. the first will flush after delay*0,
655 :			/// the second after delay1, the third after delay2, etc. If set to false, all servers will flush
656 :			/// after the same delay.
657 :			/// It returns true if the command was successful on all servers.
658 :			/// </summary>
659 :			public bool FlushAll() { return FlushAll(TimeSpan.Zero, false); }
660 :			public bool FlushAll(TimeSpan delay) { return FlushAll(delay, false); }
661 :			public bool FlushAll(TimeSpan delay, bool staggered) {
662 :			bool noerrors = true;
663 :			uint count = 0;
664 :			foreach (SocketPool pool in serverPool.HostList) {
665 :			serverPool.Execute(pool, delegate(PooledSocket socket) {
666 :			uint delaySeconds = (staggered ? (uint)delay.TotalSeconds * count : (uint)delay.TotalSeconds);
667 :			//Funnily enough, "flush_all 0" has no effect, you have to send "flush_all" to flush immediately.
668 :			socket.Write("flush_all " + (delaySeconds==0?"":delaySeconds.ToString()) + "\r\n");
669 :			if (!socket.ReadResponse().StartsWith("OK")) {
670 :			noerrors = false;
671 :			}
672 :			count++;
673 :			});
674 :			}
675 :			return noerrors;
676 :			}
677 :			#endregion
678 :
679 :			#region Stats
680 :			/// <summary>
681 :			/// This method corresponds to the "stats" command in the memcached protocol.
682 :			/// It will send the stats command to all servers, and it will return a Dictionary for each server
683 :			/// containing the results of the command.
684 :			/// </summary>
685 :			public Dictionary<string, Dictionary<string, string>> Stats() {
686 :			Dictionary<string, Dictionary<string, string>> results = new Dictionary<string, Dictionary<string, string>>();
687 :			foreach (SocketPool pool in serverPool.HostList) {
688 :			results.Add(pool.Host, stats(pool));
689 :			}
690 :			return results;
691 :			}
692 :
693 :			/// <summary>
694 :			/// This method corresponds to the "stats" command in the memcached protocol.
695 :			/// It will send the stats command to the server that corresponds to the given key, hash or host,
696 :			/// and return a Dictionary containing the results of the command.
697 :			/// </summary>
698 :			public Dictionary<string, string> Stats(string key) { return Stats(hash(key)); }
699 :			public Dictionary<string, string> Stats(uint hash) { return stats(serverPool.GetSocketPool(this.hash(hash))); }
700 :			public Dictionary<string, string> StatsByHost(string host) { return stats(serverPool.GetSocketPool(host)); }
701 :			private Dictionary<string, string> stats(SocketPool pool) {
702 :			if (pool == null) {
703 :			return null;
704 :			}
705 :			Dictionary<string, string> result = new Dictionary<string, string>();
706 :			serverPool.Execute(pool, delegate(PooledSocket socket) {
707 :			socket.Write("stats\r\n");
708 :			string line;
709 :			while (!(line = socket.ReadResponse().TrimEnd('\0', '\r', '\n')).StartsWith("END")) {
710 :			string[] s = line.Split(' ');
711 :			result.Add(s[1], s[2]);
712 :			}
713 :			});
714 :			return result;
715 :			}
716 :
717 :			#endregion
718 :
719 :			#region Status
720 :			/// <summary>
721 :			/// This method retrives the status from the serverpool. It checks the connection to all servers
722 :			/// and returns usage statistics for each server.
723 :			/// </summary>
724 :			public Dictionary<string, Dictionary<string, string>> Status() {
725 :			Dictionary<string, Dictionary<string, string>> results = new Dictionary<string, Dictionary<string, string>>();
726 :			foreach (SocketPool pool in serverPool.HostList) {
727 :			Dictionary<string, string> result = new Dictionary<string, string>();
728 :			if (serverPool.Execute<bool>(pool, false, delegate { return true; })) {
729 :			result.Add("Status", "Ok");
730 :			} else {
731 :			result.Add("Status", "Dead, next retry at: " + pool.DeadEndPointRetryTime);
732 :			}
733 :			result.Add("Sockets in pool", pool.Poolsize.ToString());
734 :			result.Add("Acquired sockets", pool.Acquired.ToString());
735 :			result.Add("Sockets reused", pool.ReusedSockets.ToString());
736 :			result.Add("New sockets created", pool.NewSockets.ToString());
737 :			result.Add("New sockets failed", pool.FailedNewSockets.ToString());
738 :			result.Add("Sockets died in pool", pool.DeadSocketsInPool.ToString());
739 :			result.Add("Sockets died on return", pool.DeadSocketsOnReturn.ToString());
740 :			result.Add("Dirty sockets on return", pool.DirtySocketsOnReturn.ToString());
741 :
742 :			results.Add(pool.Host, result);
743 :			}
744 :			return results;
745 :			}
746 :			#endregion
747 :			}
748 :			}

	ViewVC Help
Powered by ViewVC 1.0.0

MyGforge Home

Cable Beach Asset Server

Annotation of /trunk/lib/beitmemcached/MemcachedClient.cs