dispatcher.java

javaP2P技术内幕课程111213141516源代码

/*
 * Created by IntelliJ IDEA.
 * User: fsommers
 * Date: Apr 7, 2002
 * Time: 10:26:30 PM
 * To change template for new class use 
 * Code Style | Class Templates options (Tools | IDE Options).
 */
package primecruncher;

import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.discovery.DiscoveryService;
import net.jxta.discovery.DiscoveryListener;
import net.jxta.discovery.DiscoveryEvent;
import net.jxta.pipe.PipeService;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.PipeID;
import net.jxta.exception.PeerGroupException;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.ModuleSpecAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.document.*;
import net.jxta.id.IDFactory;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.codat.Codat;

import java.util.*;
import java.io.*;
import java.net.URL;
import java.net.MalformedURLException;
import java.net.UnknownServiceException;

/**
 * This class acts as a client-side proxy for the service.
 */
public class Dispatcher implements DiscoveryListener {

    private static PeerGroup group;
    private static DiscoveryService discoSvc;
    private static PipeService pipeSvc;
    private static Random random;

    //Keep cached list of peers that offer the service
    private HashSet peerCache = new HashSet();

    private OutputPipe outputPipe;

    private HashMap jobMap = new HashMap();

    public Dispatcher() {
        startJxta();
        random = new Random();
        doDiscovery();
    }



    private void doDiscovery() {

        System.out.println("Starting service discovery...");
  /*
        System.out.println("Searching local cache for " +
            ServiceConstants.SPEC_NAME + " advertisements");

        Enumeration res = null;

        try {
            res = discoSvc.getLocalAdvertisements(DiscoveryService.ADV,
                "Name", ServiceConstants.SPEC_NAME);
        } catch (IOException e) {
            System.out.println("IO Exception.");
        }
        if (res != null) {
            while (res.hasMoreElements()) {
                //this will be a ModuleSpecAdvertisement
                //adverts.add(res.nextElement());
            }
        }
    */
        System.out.println("Starting remote discovery...");
        discoSvc.getRemoteAdvertisements(null, DiscoveryService.ADV,
                "Name", ServiceConstants.SPEC_NAME, 1, this);

    }

    //create request ID. break down list into smaller lists, according to metrics on each
    //peer in the cache. Create a new dispatcher job for each peer in a new
    //thread, associate that list with the request ID.
    //resigter a result listener with each
    //each dispatcher job will make a call-back to this class, when (a)
    //it got the results, or (b) if the timeout for that peer has expired
    //Once all results are received, we call ResultListener and produce the
    //result document.
    //
    public void processPrimes(int low, int high, ResultListener listener) {

        //since pending jobs are hashed against a job id, check if no existing job
        //has the same id
        String jobID = null;
        while (jobID == null || (jobMap.containsKey(jobID))) {
            jobID = new Long(random.nextLong()).toString();
        }
        System.out.println("Allocated job id: " + jobID);
        PendingJob pendingJob = new PendingJob(jobID, this, listener);
        jobMap.put(jobID, pendingJob);

        //TODO: prune the cache here

        //create array of all known peers
        PeerInfoBundle[] peers = (PeerInfoBundle[])
                   peerCache.toArray(new PeerInfoBundle[peerCache.size()]);

        //TODO: What to do when there are no peers ready to take the job,
        //i.e., peers.length == 0



        //1. Create an empty PendingJob -- it collects all the DispatcherJobs
        // when a job is ready, it calls its pendingJob. whenver a such a call
        //arrives, PendingJob checks if all jobs are complete. If so, it calls
        //this Dispatcher's jobComplete(). JobComplete() then normalizes the results,
        //and calls result listener with the answer. A PendingJob is hashed againsta  JobID.
        System.out.println("Peers length is " + peers.length);
        //we include the highest number
        int segment = high - low + 1;
        int mod = segment % peers.length;
        int perPiece = (segment - mod)/peers.length;
        int l = low;
        int h = 0;
        int count = 1;
        for (int i=0; i < peers.length; i++) {
            h = l + perPiece -1;
            if (i == peers.length - 1) {
                h  += mod;
            }

            DispatcherJob job = new DispatcherJob(jobID, count, l, h, peers[i], pendingJob, group);
            Thread thread = new Thread(job);
            thread.setName("Dispatcher job thread " + i);
            pendingJob.addJob(job);
            thread.start();

            l += perPiece;
            count++;
        }

    }

    /**
     * All computations for a pending job have completed. We're ready.
     */
    void jobComplete(PendingJob job, Result[] subResults) {
        //normalize results, we're done,
        Result finalResult = normalizeResults(subResults);

        //call result listener with result.
        ResultListener listener = job.getResultListener();
        listener.resultEvent(finalResult);

        jobMap.remove(job.getID());
        System.out.println("Removed pending job from job map.");
        //shell we null out listener? It should destroy all subjobs as well


    }

    private Result normalizeResults(Result[] subRes) {
        //create a string for the result
        StringBuffer b = new StringBuffer();
        for (int i=0; i < subRes.length; i++) {
            Message mes = subRes[i].getMessage();
            MessageElement el = mes.getElement(ServiceConstants.PRIMELIST);
            b.append(new String(el.getBytesOffset()));

        }

        Result finRes = new Result(b.toString());
        return finRes;

    }

    private void pruneCache() {

     //do some cache management here - eliminate stale adverts

         HashSet adCopy = null;
         synchronized(peerCache) {
             adCopy = (HashSet)peerCache.clone();
         }

         //do some pruning here
         long currentTime = System.currentTimeMillis();
         Iterator it = adCopy.iterator();
         while (it.hasNext()) {
             ModuleSpecAdvertisement ad = (ModuleSpecAdvertisement)it.next();
             if (ad.getLocalExpirationTime() < currentTime + (2 * 60 * 1000)) {
                 adCopy.remove(ad);
             }
         }
         //REMIND: Is this OK?
         peerCache = adCopy;

    }



    private void startJxta() {

        try {
            group = PeerGroupFactory.newNetPeerGroup();
            discoSvc = group.getDiscoveryService();
            pipeSvc = group.getPipeService();
        } catch (PeerGroupException e) {
            System.out.println("Can't create net peer group: " + e.getMessage());
            System.exit(-1);
        }

    }

    public void discoveryEvent(DiscoveryEvent event) {
        System.out.println("DiscoveryEvent called");
        DiscoveryResponseMsg  mes = event.getResponse();

        String padv = mes.getPeerAdv();
        PeerAdvertisement peerAdv = null;
        try {
            peerAdv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(
                    new MimeMediaType("text/xml"),
                    new ByteArrayInputStream(padv.getBytes()));
        } catch (IOException ex) {
            ex.printStackTrace();
        }

        //get the ModuleSpecAdvs from this
        ModuleSpecAdvertisement moduleSpecAd = null;

        Enumeration en = mes.getResponses();
        try {
            //REMIND: can there be many???
            while (en.hasMoreElements()) {
                String st = (String)en.nextElement();
                moduleSpecAd =
                    (ModuleSpecAdvertisement)AdvertisementFactory.newAdvertisement(
                            new MimeMediaType("text/xml"),
                            new ByteArrayInputStream(st.getBytes()));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        if (peerAdv != null && moduleSpecAd != null) {
            PeerInfoBundle bundle = new PeerInfoBundle(peerAdv, moduleSpecAd);
            if (!peerCache.contains(bundle)) {
                peerCache.add(bundle);
                System.out.println("Discovered peer, added to cache");
            } else {
                System.out.println("Discovered peer, but it's already cached.");
            }

        }

    }


}