/[storm]/storm/org/nongnu/storm/modules/gispmap/GispP2PMap.java
ViewVC logotype

Diff of /storm/org/nongnu/storm/modules/gispmap/GispP2PMap.java

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.10 by hemppah, Wed May 7 10:55:10 2003 UTC revision 1.11 by benja, Sun May 11 17:33:39 2003 UTC
# Line 32  import org.nongnu.storm.impl.p2p.*; Line 32  import org.nongnu.storm.impl.p2p.*;
32  import org.nongnu.storm.util.ByteArrayKey;  import org.nongnu.storm.util.ByteArrayKey;
33    
34  import java.io.*;  import java.io.*;
35    import java.lang.ref.SoftReference;
36  import java.util.*;  import java.util.*;
37  import com.axlight.jnushare.gisp.*;  import com.axlight.jnushare.gisp.*;
38    
# Line 44  import com.axlight.jnushare.gisp.*; Line 45  import com.axlight.jnushare.gisp.*;
45   *  That's also the main reason this is in a module.   *  That's also the main reason this is in a module.
46   */   */
47  public class GispP2PMap implements P2PMap {  public class GispP2PMap implements P2PMap {
48      static public boolean dbg = false;      static public boolean dbg = true;
49      static private void pa(String s) { System.out.println(s); }      static private void pa(String s) { System.out.println(s); }
50    
51      protected GISP gisp;      protected GISP gisp;
52        protected Set entries = new HashSet();
53      protected Map cache = new HashMap();      protected Map cache = new HashMap();
54            
55      /** The "STORM" port. */      /** The "STORM" port. */
# Line 59  public class GispP2PMap implements P2PMa Line 61  public class GispP2PMap implements P2PMa
61          if(seedAddresses != null)          if(seedAddresses != null)
62              gisp.addSeedAddresses(seedAddresses);              gisp.addSeedAddresses(seedAddresses);
63          gisp.start(new String[] {"strength_min=1"});          gisp.start(new String[] {"strength_min=1"});
64    
65            // thread used to republish items into the hashtable
66            // GISP drops items after a 30 minute timeout
67            keepaliveThread.start();
68                                    
69          /** Doesn't work...maybe too quicly requested            /** Doesn't work...maybe requested too quickly
70           pa("Number of known peers in the network: " + String.valueOf(this.gisp.getNumOfPeers()));           pa("Number of known peers in the network: " + String.valueOf(this.gisp.getNumOfPeers()));
71          */                */      
72      }      }
# Line 83  public class GispP2PMap implements P2PMa Line 89  public class GispP2PMap implements P2PMa
89      public void put(String key, String value) throws IOException {      public void put(String key, String value) throws IOException {
90          gisp.insert(key, value);          gisp.insert(key, value);
91          cache.remove(key);          cache.remove(key);
92            synchronized(entries) {
93                entries.add(new Entry(key, value));
94            }
95        }
96    
97          // XXX keep alive after ttl...      /** Publish all entries from the map in the DHT again.
98         *  Used to keep them alive after timeout.
99         */
100        public void republish() {
101            synchronized(entries) {
102                for(Iterator i=entries.iterator(); i.hasNext();) {
103                    Entry e = (Entry)i.next();
104                    gisp.insert(e.key, e.value);
105                }
106            }
107      }      }
108    
109      public Collector get(String key) throws IOException {      public Collector get(String key) throws IOException {
# Line 123  public class GispP2PMap implements P2PMa Line 142  public class GispP2PMap implements P2PMa
142          }          }
143      }      }
144    
145        protected class Entry {
146            protected final String key, value;
147            protected Entry(String k, String v) { key=k; value=v; }
148        }
149    
150        protected Thread keepaliveThread = new Thread(new Keepalive(this));
151        protected static class Keepalive implements Runnable {
152            SoftReference ref;
153            protected Keepalive(GispP2PMap map) {
154                this.ref = new SoftReference(map);
155            }
156            public void run() {
157                while(true) {
158                    // wait 25 minutes before republishing everything
159                    // GISP timeout is 30 minutes
160                    try {
161                        Thread.sleep(25 * 60 * 1000);
162                    } catch(InterruptedException _) {}
163                    GispP2PMap map = (GispP2PMap)ref.get();
164                    if(map == null) break;
165                    if(dbg) pa("Republish GispP2PMap");
166                    map.republish();
167                }
168                if(dbg) pa("GispP2PMap object garbage collected");
169            }
170        }
171    
172        protected void finalize() {
173            keepaliveThread.interrupt();
174        }
175    
176      public static void main(String argv[]) throws Exception {      public static void main(String argv[]) throws Exception {
177          new GispP2PMap(PORT, argv);              new GispP2PMap(PORT, argv);    
178      }      }

Legend:
Removed from v.1.10  
changed lines
  Added in v.1.11

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26