grouprequest.java

来自「JGRoups源码」· Java 代码 · 共 616 行 · 第 1/2 页

     */
    public void viewChange(View new_view) {
        Address mbr;
        Vector mbrs=new_view != null? new_view.getMembers() : null;
        if(requests == null || requests.size() == 0 || mbrs == null)
            return;

        synchronized(this.members) {
            this.members.clear();
            this.members.addAll(mbrs);
        }

        Map.Entry entry;
        Rsp rsp;
        boolean modified=false;
        synchronized(requests) {
            for(Iterator it=requests.entrySet().iterator(); it.hasNext();) {
                entry=(Map.Entry)it.next();
                mbr=(Address)entry.getKey();
                if(!mbrs.contains(mbr)) {
                    addSuspect(mbr);
                    rsp=(Rsp)entry.getValue();
                    rsp.setValue(null);
                    rsp.setSuspected(true);
                    modified=true;
                }
            }
            if(modified)
                requests.notifyAll();
        }
    }


    /* -------------------- End of Interface RspCollector ----------------------------------- */



    /** Returns the results as a RspList */
    public RspList getResults() {
        synchronized(requests) {
            Collection rsps=requests.values();
            return new RspList(rsps);
        }
    }


    public String toString() {
        StringBuffer ret=new StringBuffer(128);
        ret.append("[GroupRequest:\n");
        ret.append("req_id=").append(req_id).append('\n');
        if(caller != null)
            ret.append("caller=").append(caller).append("\n");

        Map.Entry entry;
        Address mbr;
        Rsp rsp;
        synchronized(requests) {
            for(Iterator it=requests.entrySet().iterator(); it.hasNext();) {
                entry=(Map.Entry)it.next();
                mbr=(Address)entry.getKey();
                rsp=(Rsp)entry.getValue();
                ret.append(mbr).append(": ").append(rsp).append("\n");
            }
        }
        if(suspects.size() > 0)
            ret.append("\nsuspects: ").append(suspects);
        ret.append("\nrequest_msg: ").append(request_msg);
        ret.append("\nrsp_mode: ").append(modeToString(rsp_mode));
        ret.append("\ndone: ").append(done);
        ret.append("\ntimeout: ").append(timeout);
        ret.append("\nexpected_mbrs: ").append(expected_mbrs);
        ret.append("\n]");
        return ret.toString();
    }


    public int getNumSuspects() {
        return suspects.size();
    }


    public Vector getSuspects() {
        return suspects;
    }


    public boolean isDone() {
        return done;
    }



    /* --------------------------------- Private Methods -------------------------------------*/

    private int determineMajority(int i) {
        return i < 2? i : (i / 2) + 1;
    }

    /** Generates a new unique request ID */
    private static synchronized long getRequestId() {
        long result=System.currentTimeMillis();
        if(result <= last_req_id) {
            result=last_req_id + 1;
        }
        last_req_id=result;
        return result;
    }

    /** This method runs with rsp_mutex locked (called by <code>execute()</code>). */
    private boolean doExecute(long timeout) throws Exception {
        long start_time=0;
        Address suspect;
        req_id=getRequestId();
        reset(null); // clear 'responses' array

        synchronized(requests) {
            for(int i=0; i < suspects.size(); i++) {  // mark all suspects in 'received' array
                suspect=(Address)suspects.elementAt(i);
                Rsp rsp=(Rsp)requests.get(suspect);
                if(rsp != null) {
                    rsp.setSuspected(true);
                    break; // we can break here because we ensure there are no duplicate members
                }
            }
        }

        try {
            if(log.isTraceEnabled()) log.trace(new StringBuffer("sending request (id=").append(req_id).append(')'));
            if(corr != null) {
                java.util.List tmp=new Vector(members);
                corr.sendRequest(req_id, tmp, request_msg, rsp_mode == GET_NONE? null : this);
            }
            else {
                transport.send(request_msg);
            }
        }
        catch(Exception ex) {
            if(corr != null)
                corr.done(req_id);
            throw ex;
        }

        synchronized(requests) {
            if(timeout <= 0) {
                while(true) { /* Wait for responses: */
                    adjustMembership(); // may not be necessary, just to make sure...
                    if(responsesComplete()) {
                        if(corr != null) {
                            corr.done(req_id);
                        }
                        if(log.isTraceEnabled()) {
                            log.trace("received all responses: " + toString());
                        }
                        return true;
                    }
                    try {
                        requests.wait();
                    }
                    catch(Exception e) {
                    }
                }
            }
            else {
                start_time=System.currentTimeMillis();
                while(timeout > 0) { /* Wait for responses: */
                    if(responsesComplete()) {
                        if(corr != null)
                            corr.done(req_id);
                        if(log.isTraceEnabled()) log.trace("received all responses: " + toString());
                        return true;
                    }
                    timeout=timeout - (System.currentTimeMillis() - start_time);
                    if(timeout > 0) {
                        try {
                            requests.wait(timeout);
                        }
                        catch(Exception e) {
                        }
                    }
                }
                if(corr != null) {
                    corr.done(req_id);
                }
                return false;
            }
        }
    }

    private boolean responsesComplete() {
        int num_received=0, num_not_received=0, num_suspected=0;
        final int num_total=requests.size();

        Rsp rsp;
        for(Iterator it=requests.values().iterator(); it.hasNext();) {
            rsp=(Rsp)it.next();
            if(rsp.wasReceived()) {
                num_received++;
            }
            else {
                if(rsp.wasSuspected()) {
                    num_suspected++;
                }
                else {
                    num_not_received++;
                }
            }
        }

        switch(rsp_mode) {
            case GET_FIRST:
                if(num_received > 0)
                    return true;
                if(num_suspected >= num_total)
                // e.g. 2 members, and both suspected
                    return true;
                break;
            case GET_ALL:
                return num_received + num_suspected >= num_total;
            case GET_MAJORITY:
                int majority=determineMajority(num_total);
                if(num_received + num_suspected >= majority)
                    return true;
                break;
            case GET_ABS_MAJORITY:
                majority=determineMajority(num_total);
                if(num_received >= majority)
                    return true;
                break;
            case GET_N:
                if(expected_mbrs >= num_total) {
                    rsp_mode=GET_ALL;
                    return responsesComplete();
                }
                if(num_received >= expected_mbrs) {
                    return true;
                }
                if(num_received + num_not_received < expected_mbrs) {
                    return num_received + num_suspected >= expected_mbrs;
                }
                return false;
            case GET_NONE:
                return true;
            default :
                if(log.isErrorEnabled()) log.error("rsp_mode " + rsp_mode + " unknown !");
                break;
        }
        return false;
    }





    /**
     * Adjusts the 'received' array in the following way:
     * <ul>
     * <li>if a member P in 'membership' is not in 'members', P's entry in the 'received' array
     *     will be marked as SUSPECTED
     * <li>if P is 'suspected_mbr', then P's entry in the 'received' array will be marked
     *     as SUSPECTED
     * </ul>
     * This call requires exclusive access to rsp_mutex (called by getResponses() which has
     * a the rsp_mutex locked, so this should not be a problem).
     */
    private void adjustMembership() {
        if(requests.size() == 0)
            return;

        Map.Entry entry;
        Address mbr;
        Rsp rsp;
        for(Iterator it=requests.entrySet().iterator(); it.hasNext();) {
            entry=(Map.Entry)it.next();
            mbr=(Address)entry.getKey();
            if((!this.members.contains(mbr)) || suspects.contains(mbr)) {
                addSuspect(mbr);
                rsp=(Rsp)entry.getValue();
                rsp.setValue(null);
                rsp.setSuspected(true);
            }
        }
    }

    /**
     * Adds a member to the 'suspects' list. Removes oldest elements from 'suspects' list
     * to keep the list bounded ('max_suspects' number of elements)
     */
    private void addSuspect(Address suspected_mbr) {
        if(!suspects.contains(suspected_mbr)) {
            suspects.addElement(suspected_mbr);
            while(suspects.size() >= max_suspects && suspects.size() > 0)
                suspects.remove(0); // keeps queue bounded
        }
    }

    private String modeToString(int m) {
        switch(m) {
            case GET_FIRST: return "GET_FIRST";
            case GET_ALL: return "GET_ALL";
            case GET_MAJORITY: return "GET_MAJORITY";
            case GET_ABS_MAJORITY: return "GET_ABS_MAJORITY";
            case GET_N: return "GET_N";
            case GET_NONE: return "GET_NONE";
            default: return "<unknown> (" + m + ")";
        }
    }
}