entropy.cache

25 """ 26 Entropy asynchronous and synchronous cache writer 27 and reader. This class is a Singleton and contains 28 a thread doing the cache writes asynchronously, thus 29 it must be stopped before your application is terminated 30 calling the stop() method. 31 32 Sample code: 33 34 >>> # import module 35 >>> from entropy.cache import EntropyCacher 36 ... 37 >>> # first EntropyCacher load, start it 38 >>> cacher = EntropyCacher() 39 >>> cacher.start() 40 ... 41 >>> # now store something into its cache 42 >>> cacher.push('my_identifier1', [1, 2, 3]) 43 >>> # now store something synchronously 44 >>> cacher.push('my_identifier2', [1, 2, 3], async = False) 45 ... 46 >>> # now flush all the caches to disk, and make sure all 47 >>> # is written 48 >>> cacher.sync(wait = True) 49 ... 50 >>> # now fetch something from the cache 51 >>> data = cacher.pop('my_identifier1') 52 [1, 2, 3] 53 ... 54 >>> # now discard all the cached (async) writes 55 >>> cacher.discard() 56 ... 57 >>> # and stop EntropyCacher 58 >>> cacher.stop() 59 60 """ 61 62 import entropy.dump as dumpTools 63 import entropy.tools as entropyTools 64 import copy 65

66 - def init_singleton(self):

67 """ 68 Singleton overloaded method. Equals to __init__. 69 This is the place where all the properties initialization 70 takes place. 71 """ 72 import threading 73 self.__alive = False 74 self.__cache_writer = None 75 self.__cache_buffer = Lifo() 76 self.__cache_lock = threading.Lock()

77

78 - def __copy_obj(self, obj):

79 """ 80 Return a copy of an object done by the standard 81 library "copy" module. 82 83 @param obj: object to copy 84 @type obj: any Python object 85 @rtype: copied object 86 @return: copied object 87 """ 88 return self.copy.deepcopy(obj)

89

90 - def __cacher(self):

91 """ 92 This is where the actual asynchronous copy takes 93 place. __cacher runs on a different threads and 94 all the operations done by this are atomic and 95 thread-safe. It just loops over and over until 96 __alive becomes False. 97 """ 98 while 1: 99 if not self.__alive: 100 break 101 with self.__cache_lock: 102 try: 103 data = self.__cache_buffer.pop() 104 except (ValueError, TypeError,): 105 # TypeError is when objects are being destroyed 106 break # stack empty 107 key, data = data 108 d_o = self.dumpTools.dumpobj 109 if not d_o: 110 break 111 d_o(key, data)

112

113 - def __del__(self):

114 self.stop()

115

116 - def start(self):

117 """ 118 This is the method used to start the asynchronous cache 119 writer but also the whole cacher. If this method is not 120 called, the instance will always trash and cache write 121 request. 122 123 @return: None 124 """ 125 with self.__cache_lock: 126 self.__cache_buffer.clear() 127 self.__cache_writer = TimeScheduled(1, self.__cacher) 128 self.__cache_writer.set_delay_before(True) 129 self.__cache_writer.start() 130 while not self.__cache_writer.isAlive(): 131 continue 132 self.__alive = True

133

134 - def is_started(self):

135 """ 136 Return whether start is called or not. This equals to 137 checking if the cacher is running, thus is writing cache 138 to disk. 139 140 @return: None 141 """ 142 return self.__alive

143

144 - def stop(self):

145 """ 146 This method stops the execution of the cacher, which won't 147 accept cache writes anymore. The thread responsible of writing 148 to disk is stopped here and the Cacher will be back to being 149 inactive. A watchdog will avoid the thread to freeze the 150 call if the write buffer is overloaded. 151 152 @return: None 153 """ 154 155 watch_dog = 80 156 while self.__cache_buffer.is_filled() and (watch_dog > 0): 157 watch_dog -= 1 158 time.sleep(0.125) 159 self.__alive = False 160 161 with self.__cache_lock: 162 self.__cache_buffer.clear() 163 if self.__cache_writer != None: 164 self.__cache_writer.kill() 165 self.__cache_writer.join() 166 self.__cache_writer = None

167

168 - def sync(self, wait = False):

169 """ 170 This method can be called anytime and forces the instance 171 to flush all the cache writes queued to disk. If wait == False 172 a watchdog prevents this call to get stuck in case of write 173 buffer overloads. 174 175 @keyword wait: indicates if waiting until done (synchronous mode) or not 176 @type wait: bool 177 @rtype: None 178 @return: None 179 """ 180 if not self.__alive: 181 self.__cache_buffer.clear() 182 return 183 184 watch_dog = 40 185 while self.__cache_buffer.is_filled() and ((watch_dog > 0) or wait) \ 186 and self.__alive: 187 188 if not wait: 189 watch_dog -= 1 190 time.sleep(0.125) 191 192 self.__cache_buffer.clear()

193

194 - def discard(self):

195 """ 196 This method makes buffered cache to be discarded synchronously. 197 198 @return: None 199 """ 200 self.__cache_buffer.clear() 201 with self.__cache_lock: 202 self.__cache_buffer.clear() # make sure twice

203

204 - def push(self, key, data, async = True):

205 """ 206 This is the place where data is either added 207 to the write queue or written to disk (if async == False) 208 only and only if start() method has been called. 209 210 @param key: cache data identifier 211 @type key: string 212 @param data: picklable object 213 @type data: any picklable object 214 @keyword async: store cache asynchronously or not 215 @type async: bool 216 @rtype: None 217 @return: None 218 """ 219 if not self.__alive: 220 return 221 if async: 222 with self.__cache_lock: 223 self.__cache_buffer.push((key, self.__copy_obj(data),)) 224 else: 225 self.dumpTools.dumpobj(key, data)

226

227 - def pop(self, key):

228 """ 229 This is the place where data is retrieved from cache. 230 You must know the cache identifier used when push() 231 was called. 232 233 @param key: cache data identifier 234 @type key: string 235 @rtype: Python object 236 @return: object stored into the stack or None (if stack is empty) 237 """ 238 with self.__cache_lock: 239 l_o = self.dumpTools.loadobj 240 if not l_o: 241 return 242 return l_o(key)

Source Code for Module entropy.cache