Java > Open Source Codes > com > caucho > server > cluster > StoreManager


1 /*
2  * Copyright (c) 1998-2006 Caucho Technology -- all rights reserved
3  *
4  * This file is part of Resin(R) Open Source
5  *
6  * Each copy or derived work must preserve the copyright notice and this
7  * notice unmodified.
8  *
9  * Resin Open Source is free software; you can redistribute it and/or modify
10  * it under the terms of the GNU General Public License as published by
11  * the Free Software Foundation; either version 2 of the License, or
12  * (at your option) any later version.
13  *
14  * Resin Open Source is distributed in the hope that it will be useful,
15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, or any warranty
17  * of NON-INFRINGEMENT. See the GNU General Public License for more
18  * details.
19  *
20  * You should have received a copy of the GNU General Public License
21  * along with Resin Open Source; if not, write to the
22  *
23  * Free Software Foundation, Inc.
24  * 59 Temple Place, Suite 330
25  * Boston, MA 02111-1307 USA
26  *
27  * @author Scott Ferguson
28  */
29
30 package com.caucho.server.cluster;
31
32 import com.caucho.config.types.Period;
33 import com.caucho.lifecycle.Lifecycle;
34 import com.caucho.loader.ClassLoaderListener;
35 import com.caucho.loader.DynamicClassLoader;
36 import com.caucho.loader.Environment;
37 import com.caucho.loader.EnvironmentClassLoader;
38 import com.caucho.loader.EnvironmentListener;
39 import com.caucho.management.server.PersistentStoreMXBean;
40 import com.caucho.util.Alarm;
41 import com.caucho.util.AlarmListener;
42 import com.caucho.util.L10N;
43 import com.caucho.util.Log;
44 import com.caucho.util.LruCache;
45 import com.caucho.vfs.TempStream;
46
47 import javax.annotation.PostConstruct;
48 import java.io.IOException ;
49 import java.util.HashMap ;
50 import java.util.logging.Level ;
51 import java.util.logging.Logger ;
52
53 /**
54  * Base class for distributed stores.
55  */
56 abstract public class StoreManager
57   implements AlarmListener, EnvironmentListener, ClassLoaderListener {
58   static protected final Logger log = Log.open(StoreManager.class);
59   static final L10N L = new L10N(StoreManager.class);
60   
61   private static int DECODE[];
62
63   private Cluster _cluster;
64   private String _serverId;
65   
66   protected int _selfIndex;
67   private ServerConnector []_serverList;
68   
69   private Alarm _alarm;
70   
71   // protected long _maxIdleTime = 24 * 3600 * 1000L;
72 protected long _maxIdleTime = 3 * 3600 * 1000L;
73   protected long _idleCheckInterval = 5 * 60 * 1000L;
74
75   protected boolean _isAlwaysLoad;
76   protected boolean _isAlwaysSave;
77   
78   protected HashMap <String ,Store> _storeMap;
79   protected LruCache<String ,ClusterObject> _clusterObjects
80     = new LruCache<String ,ClusterObject>(4096);
81
82   private final Lifecycle _lifecycle = new Lifecycle(log, toString());
83
84   //
85 // statistics
86 //
87
88   protected volatile long _loadCount;
89   protected volatile long _loadFailCount;
90
91   protected volatile long _saveCount;
92   protected volatile long _saveFailCount;
93
94   protected StoreManager()
95   {
96     _storeMap = new HashMap <String ,Store>();
97
98     _alarm = new Alarm(this);
99
100     Environment.addClassLoaderListener(this);
101   }
102
103   /**
104    * Sets the cluster.
105    */
106   public void setCluster(Cluster cluster)
107   {
108     _cluster = cluster;
109   }
110
111   /**
112    * Gets the cluster.
113    */
114   public Cluster getCluster()
115   {
116     return _cluster;
117   }
118
119   /**
120    * Returns the admin.
121    */
122   public PersistentStoreMXBean getAdmin()
123   {
124     return null;
125   }
126
127   /**
128    * Set true if the store should always try to load the object.
129    */
130   public void setAlwaysLoad(boolean alwaysLoad)
131   {
132     _isAlwaysLoad = alwaysLoad;
133   }
134
135   /**
136    * Set true if the store should always try to load the object.
137    */
138   public boolean isAlwaysLoad()
139   {
140     return _isAlwaysLoad;
141   }
142
143   /**
144    * Set true if the store should always try to store the object.
145    */
146   public void setAlwaysSave(boolean alwaysSave)
147   {
148     _isAlwaysSave = alwaysSave;
149   }
150
151   /**
152    * Set true if the store should always try to store the object.
153    */
154   public boolean isAlwaysSave()
155   {
156     return _isAlwaysSave;
157   }
158
159   /**
160    * Returns the length of time an idle object can remain in the store before
161    * being cleaned.
162    */
163   public long getMaxIdleTime()
164   {
165     return _maxIdleTime;
166   }
167
168   /**
169    * Sets the length of time an idle object can remain in the store before
170    * being cleaned.
171    */
172   public void setMaxIdleTime(Period maxIdleTime)
173   {
174     _maxIdleTime = maxIdleTime.getPeriod();
175   }
176
177   /**
178    * Sets the idle check interval for the alarm.
179    */
180   public void updateIdleCheckInterval(long idleCheckInterval)
181   {
182     if (_idleCheckInterval > 0 && idleCheckInterval > 0 &&
183     idleCheckInterval < _idleCheckInterval) {
184       _idleCheckInterval = idleCheckInterval;
185       _alarm.queue(idleCheckInterval);
186     }
187
188     if (_idleCheckInterval >= 0 && _idleCheckInterval < 1000)
189       _idleCheckInterval = 1000;
190   }
191
192   /**
193    * Sets the idle check interval for the alarm.
194    */
195   public long getIdleCheckTime()
196   {
197     if (_idleCheckInterval > 0)
198       return _idleCheckInterval;
199     else
200       return _maxIdleTime;
201   }
202
203   /**
204    * Returns the length of time an idle object can remain in the store before
205    * being cleaned.
206    */
207   public long getAccessWindowTime()
208   {
209     long window = _maxIdleTime / 4;
210
211     if (window < 60000L)
212       return 60000L;
213     else
214       return window;
215   }
216
217   //
218 // statistics
219 //
220
221   /**
222    * Returns the objects in the store
223    */
224   public long getObjectCount()
225   {
226     return -1;
227   }
228
229   /**
230    * Returns the total objects loaded.
231    */
232   public long getLoadCount()
233   {
234     return _loadCount;
235   }
236
237   /**
238    * Returns the objects which failed to load.
239    */
240   public long getLoadFailCount()
241   {
242     return _loadFailCount;
243   }
244
245   /**
246    * Returns the total objects saved.
247    */
248   public long getSaveCount()
249   {
250     return _saveCount;
251   }
252
253   /**
254    * Returns the objects which failed to save.
255    */
256   public long getSaveFailCount()
257   {
258     return _saveFailCount;
259   }
260
261   /**
262    * Creates a Store. The Store manages
263    * persistent objects for a particular domain, like the sesions
264    * for the /foo URL.
265    *
266    * @param storeId the persistent domain.
267    */
268   public Store createStore(String storeId, ObjectManager objectManager)
269   {
270     Store store = getStore(storeId);
271
272     store.setObjectManager(objectManager);
273
274     return store;
275   }
276
277   /**
278    * Removes a Store. The Store manages
279    * persistent objects for a particular domain, like the sesions
280    * for the /foo URL.
281    *
282    * @param storeId the persistent domain.
283    */
284   public Store removeStore(String storeId)
285   {
286     Store store = getStore(storeId);
287
288     store.setObjectManager(null);
289
290     return store;
291   }
292
293   /**
294    * Creates a ClusterObjectManager. The ClusterObjectManager manages
295    * persistent objects for a particular domain, like the sesions
296    * for the /foo URL.
297    *
298    * @param storeId the persistent domain.
299    */
300   public Store getStore(String storeId)
301   {
302     synchronized (_storeMap) {
303       Store store = _storeMap.get(storeId);
304       if (store == null) {
305     store = new Store(storeId, this);
306     _storeMap.put(storeId, store);
307       }
308
309       return store;
310     }
311   }
312
313   /**
314    * Called after any factory settings.
315    */
316   @PostConstruct
317   public boolean init()
318     throws Exception
319   {
320     if (! _lifecycle.toInit())
321       return false;
322
323     _lifecycle.setName(toString());
324
325     if (_cluster == null)
326       _cluster = Cluster.getLocal();
327     
328     if (_cluster != null) {
329       _serverId = _cluster.getServerId();
330       ClusterServer selfServer = _cluster.getSelfServer();
331
332       if (selfServer != null)
333     _selfIndex = selfServer.getIndex();
334       else if (_cluster.getServerList().length > 1) {
335     // XXX: error?
336 log.warning(L.l("cluster-store for '{0}' needs an <srun> configuration for it.",
337             _serverId));
338       }
339
340       ClusterServer []serverList = _cluster.getServerList();
341       
342       _serverList = new ServerConnector[serverList.length];
343
344       for (int i = 0; i < serverList.length; i++) {
345     _serverList[i] = serverList[i].getServerConnector();
346       }
347     }
348
349
350     Environment.addEnvironmentListener(this);
351
352     return true;
353   }
354
355   /**
356    * Called to start the store.
357    */
358   public boolean start()
359     throws Exception
360   {
361     if (! _lifecycle.toActive())
362       return false;
363     
364     // notify the siblings that we're awake
365 if (_serverList != null) {
366       ServerConnector []serverList = _serverList;
367
368       for (int i = 0; i < serverList.length; i++) {
369     ServerConnector server = serverList[i];
370
371     if (server == null)
372       continue;
373
374     ClusterClient client = server.getClient();
375
376     try {
377       if (client != null) {
378         ClusterStream s = client.open();
379         s.close();
380       }
381     } catch (Throwable e) {
382     }
383       }
384       
385     }
386
387     handleAlarm(_alarm);
388
389     return true;
390   }
391
392   /**
393    * Cleans old objects. Living objects corresponding to the old
394    * objects are not cleared, since their timeout should be less than
395    * the store timeout.
396    */
397   public void clearOldObjects()
398     throws Exception
399   {
400   }
401
402   /**
403    * Returns true if this server is a primary for the given object id.
404    */
405   protected boolean isPrimary(String id)
406   {
407     return getPrimaryIndex(id, 0) == getSelfIndex();
408   }
409
410   /**
411    * Returns the owning index.
412    */
413   public int getPrimaryIndex(String id, int offset)
414   {
415     return 0;
416   }
417
418   /**
419    * Returns the backup index.
420    */
421   public int getSecondaryIndex(String id, int offset)
422   {
423     return 0;
424   }
425
426   /**
427    * Returns the backup index.
428    */
429   public int getTertiaryIndex(String id, int offset)
430   {
431     return 0;
432   }
433
434   /**
435    * Loads object access time.
436    *
437    * @param obj the object to update.
438    */
439   abstract protected boolean load(ClusterObject clusterObject, Object obj)
440     throws Exception ;
441
442   /**
443    * Updates the object's access time.
444    *
445    * @param storeId the identifier of the storage group
446    * @param obj the object to update.
447    */
448   public void access(String uniqueId)
449     throws Exception
450   {
451     ClusterObject obj = getClusterObject(uniqueId);
452
453     if (obj != null)
454       obj.access();
455     else
456       accessImpl(uniqueId);
457   }
458
459   /**
460    * Updates the object's access time.
461    *
462    * @param storeId the identifier of the storage group
463    * @param obj the object to update.
464    */
465   public void access(Store store, String id)
466     throws Exception
467   {
468     getClusterObject(store, id).access();
469   }
470   
471   /**
472    * Updates the object's access time in the persistent store.
473    *
474    * @param uniqueId the identifier of the object.
475    */
476   abstract public void accessImpl(String uniqueId)
477     throws Exception ;
478   
479   /**
480    * Sets the timef for the expires interval.
481    *
482    * @param uniqueId the identifier of the object.
483    * @param long the time in ms for the expire
484    */
485   public void setExpireInterval(String uniqueId, long expires)
486     throws Exception
487   {
488   }
489
490   /**
491    * When the object is no longer valid, remove it from the backing store.
492    *
493    * @param storeId the identifier of the storeage group
494    * @param objectId the identifier of the object to remove
495    */
496   public void update(String storeId, String objectId)
497     throws Exception
498   {
499     ClusterObject obj = getClusterObject(storeId, objectId);
500
501     if (obj != null)
502       obj.update();
503   }
504
505   /**
506    * Updates the owner object.
507    *
508    * @param uniqueId the identifier of the storage group
509    */
510   public void updateOwner(String objectId, String uniqueId)
511     throws Exception
512   {
513     
514   }
515   
516   /**
517    * Saves the object to the cluster.
518    *
519    * @param storeId the identifier of the storage group
520    * @param obj the object to store.
521    */
522   public void store(Store store, String id, Object obj)
523     throws IOException
524   {
525     ClusterObject clusterObj = getClusterObject(store, id);
526
527     if (clusterObj != null) {
528     }
529     else if (store.getObjectManager().isEmpty(obj))
530       return;
531     else
532       clusterObj = createClusterObject(store, id);
533     
534     clusterObj.store(obj);
535   }
536
537   /**
538    * Returns the cluster object.
539    *
540    * @param storeId the identifier of the storage group
541    * @param obj the object to store.
542    */
543   ClusterObject createClusterObject(Store store, String id)
544   {
545     try {
546       String uniqueId = store.getId() + ';' + id;
547
548       synchronized (this) {
549     ClusterObject clusterObj = _clusterObjects.get(uniqueId);
550     if (clusterObj == null) {
551       clusterObj = create(store, id);
552       _clusterObjects.put(clusterObj.getUniqueId(), clusterObj);
553     }
554
555     return clusterObj;
556       }
557     } catch (Throwable e) {
558       log.log(Level.WARNING, e.toString(), e);
559
560       return null;
561     }
562   }
563
564   /**
565    * Returns the cluster object.
566    *
567    * @param storeId the identifier of the storage group
568    * @param obj the object to store.
569    */
570   ClusterObject getClusterObject(Store store, String id)
571   {
572     return getClusterObject(makeUniqueId(store, id));
573   }
574
575   /**
576    * Returns the cluster object.
577    *
578    * @param storeId the identifier of the storage group
579    * @param obj the object to store.
580    */
581   ClusterObject getClusterObject(String storeId, String id)
582   {
583     return getClusterObject(makeUniqueId(storeId, id));
584   }
585
586   /**
587    * Returns the cluster object.
588    *
589    * @param storeId the identifier of the storage group
590    * @param obj the object to store.
591    */
592   ClusterObject getClusterObject(String uniqueId)
593   {
594     return _clusterObjects.get(uniqueId);
595   }
596
597   /**
598    * Returns the cluster object.
599    *
600    * @param storeId the identifier of the storage group
601    * @param obj the object to store.
602    */
603   ClusterObject removeClusterObject(String storeId, String id)
604   {
605     synchronized (this) {
606       return _clusterObjects.remove(makeUniqueId(storeId, id));
607     }
608   }
609   
610   /**
611    * Creates the cluster object.
612    */
613   ClusterObject create(Store store, String id)
614   {
615     return new ClusterObject(this, store, id);
616   }
617   
618   /**
619    * Save the object to the store.
620    *
621    * @param storeId the identifier of the storage group
622    * @param obj the object to store.
623    */
624   abstract protected void store(ClusterObject clusterObject,
625                 TempStream tempStream,
626                 long crc)
627     throws Exception ;
628   
629   /**
630    * Handles a callback from an alarm, scheduling the timeout.
631    */
632   public void handleAlarm(Alarm alarm)
633   {
634     if (! _lifecycle.isActive())
635       return;
636
637     try {
638       clearOldObjects();
639     } catch (Throwable e) {
640       log.log(Level.WARNING, e.toString(), e);
641     } finally {
642       _alarm.queue(getIdleCheckTime());
643     }
644   }
645
646   /**
647    * When the object is no longer valid, remove it from the backing store.
648    *
649    * @param obj the object to remove
650    */
651   public void remove(ClusterObject obj)
652     throws Exception
653   {
654     
655   }
656
657   /**
658    * When the object is no longer valid, remove it from the backing store.
659    *
660    * @param store the identifier of the storeage group
661    * @param objectId the identifier of the object to remove
662    */
663   public void remove(Store store, String objectId)
664     throws Exception
665   {
666   }
667
668   /**
669    * Returns the unique id.
670    */
671   private String makeUniqueId(Store store, String objectId)
672   {
673     return store.getId() + ';' + objectId;
674   }
675
676   /**
677    * Returns the unique id.
678    */
679   private String makeUniqueId(String storeId, String objectId)
680   {
681     return storeId + ';' + objectId;
682   }
683
684   /**
685    * Returns the self servers.
686    */
687   protected int getSelfIndex()
688   {
689     return _selfIndex;
690   }
691
692   /**
693    * Returns the list of cluster servers.
694    */
695   protected ServerConnector []getServerList()
696   {
697     return _serverList;
698   }
699
700   /**
701    * Returns the cluster server which owns the object
702    */
703   protected ServerConnector getOwningServer(String objectId)
704   {
705     if (_cluster == null)
706       return null;
707     
708     char ch = objectId.charAt(0);
709     
710     ServerConnector []serverList = _serverList;
711
712     if (serverList.length > 0) {
713       int srunIndex = decode(ch) % serverList.length;
714
715       return serverList[srunIndex];
716     }
717     else
718       return null;
719   }
720   
721   /**
722    * Handles the case where the environment is activated.
723    */
724   public void environmentStart(EnvironmentClassLoader loader)
725   {
726     try {
727       start();
728     } catch (Exception e) {
729       log.log(Level.WARNING, e.toString(), e);
730     }
731   }
732   
733   /**
734    * Handles the case where the environment loader is stops
735    */
736   public void environmentStop(EnvironmentClassLoader loader)
737   {
738   }
739   
740   /**
741    * Handles the case where the environment is activated.
742    */
743   public void classLoaderInit(DynamicClassLoader loader)
744   {
745   }
746   
747   /**
748    * Handles the case where the environment loader is dropped.
749    */
750   public void classLoaderDestroy(DynamicClassLoader loader)
751   {
752     destroy();
753   }
754   
755   /**
756    * Called at end of life.
757    */
758   public void destroy()
759   {
760     if (! _lifecycle.toDestroy())
761       return;
762     
763     _alarm.dequeue();
764   }
765
766   static class ObjectKey {
767     private String _storeId;
768     private String _objectId;
769
770     ObjectKey()
771     {
772     }
773     
774     ObjectKey(String storeId, String objectId)
775     {
776       init(storeId, objectId);
777     }
778
779     void init(String storeId, String objectId)
780     {
781       _storeId = storeId;
782       _objectId = objectId;
783     }
784
785     public int hashCode()
786     {
787       return _storeId.hashCode() * 65521 + _objectId.hashCode();
788     }
789
790     public boolean equals(Object o)
791     {
792       if (this == o)
793     return true;
794       else if (! (o instanceof ObjectKey))
795     return false;
796
797       ObjectKey key = (ObjectKey) o;
798
799       return _objectId.equals(key._objectId) && _storeId.equals(key._storeId);
800     }
801   }
802
803   static int decode(int code)
804   {
805     return DECODE[code & 0x7f];
806   }
807
808   /**
809    * Converts an integer to a printable character
810    */
811   private static char convert(long code)
812   {
813     code = code & 0x3f;
814     
815     if (code < 26)
816       return (char) ('a' + code);
817     else if (code < 52)
818       return (char) ('A' + code - 26);
819     else if (code < 62)
820       return (char) ('0' + code - 52);
821     else if (code == 62)
822       return '_';
823     else
824       return '-';
825   }
826
827   static {
828     DECODE = new int[128];
829     for (int i = 0; i < 64; i++)
830       DECODE[(int) convert(i)] = i;
831   }
832 }
833

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags