requestcorrelator.java

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

     * Called by the protocol below when a message has been received. The
     * algorithm should test whether the message is destined for us and,
     * if not, pass it up to the next layer. Otherwise, it should remove
     * the header and check whether the message is a request or response.
     * In the first case, the message will be delivered to the request
     * handler registered (calling its <code>handle()</code> method), in the
     * second case, the corresponding response collector is looked up and
     * the message delivered.
     */
    public void receive(Event evt) {
        switch(evt.getType()) {
        case Event.SUSPECT:     // don't wait for responses from faulty members
            receiveSuspect((Address)evt.getArg());
            break;
        case Event.VIEW_CHANGE: // adjust number of responses to wait for
            receiveView((View)evt.getArg());
            break;

        case Event.SET_LOCAL_ADDRESS:
            setLocalAddress((Address)evt.getArg());
            break;
        case Event.MSG:
            if(!receiveMessage((Message)evt.getArg()))
                return;
            break;
        }
        if(transport instanceof Protocol)
            ((Protocol)transport).passUp(evt);
        else
            if(log.isErrorEnabled()) log.error("we do not pass up messages via Transport");
    }


    /**
     */
    public final void start() {
        if(deadlock_detection) {
            startScheduler();
        }
        started=true;
    }

    public void stop() {
        stopScheduler();
        started=false;
    }


    void startScheduler() {
        if(scheduler == null) {
            scheduler=new Scheduler();
            if(deadlock_detection && call_stack_setter == null) {
                call_stack_setter=new CallStackSetter();
                scheduler.setListener(call_stack_setter);
            }
            if(concurrent_processing)
                scheduler.setConcurrentProcessing(concurrent_processing);
            scheduler.start();
        }
    }


    void stopScheduler() {
        if(scheduler != null) {
            scheduler.stop();
            scheduler=null;
        }
    }


    // .......................................................................



    /**
     * <tt>Event.SUSPECT</tt> event received from a layer below.
     * <p>
     * All response collectors currently registered will
     * be notified that <code>mbr</code> may have crashed, so they won't
     * wait for its response.
     */
    public void receiveSuspect(Address mbr) {
        RequestEntry entry;
        // ArrayList    copy;

        if(mbr == null) return;
        if(log.isDebugEnabled()) log.debug("suspect=" + mbr);

        // copy so we don't run into bug #761804 - Bela June 27 2003
        // copy=new ArrayList(requests.values()); // removed because ConcurrentReaderHashMap can tolerate concurrent mods (bela May 8 2006)
        for(Iterator it=requests.values().iterator(); it.hasNext();) {
            entry=(RequestEntry)it.next();
            if(entry.coll != null)
                entry.coll.suspect(mbr);
        }
    }


    /**
     * <tt>Event.VIEW_CHANGE</tt> event received from a layer below.
     * <p>
     * Mark all responses from members that are not in new_view as
     * NOT_RECEIVED.
     *
     */
    public void receiveView(View new_view) {
        RequestEntry entry;
        // ArrayList    copy;

        // copy so we don't run into bug #761804 - Bela June 27 2003
        // copy=new ArrayList(requests.values());  // removed because ConcurrentReaderHashMap can tolerate concurrent mods (bela May 8 2006)
        for(Iterator it=requests.values().iterator(); it.hasNext();) {
            entry=(RequestEntry)it.next();
            if(entry.coll != null)
                entry.coll.viewChange(new_view);
        }
    }


    /**
     * Handles a message coming from a layer below
     *
     * @return true if the event should be forwarded further up, otherwise false (message was consumed)
     */
    public boolean receiveMessage(Message msg) {
        Object tmpHdr;

        // i. If header is not an instance of request correlator header, ignore
        //
        // ii. Check whether the message was sent by a request correlator with
        // the same name (there may be multiple request correlators in the same
        // protocol stack...)
        tmpHdr=msg.getHeader(name);
        if(tmpHdr == null || !(tmpHdr instanceof Header)) {
            return true;
        }

        Header hdr=(Header)tmpHdr;
        if(hdr.corrName == null || !hdr.corrName.equals(name)) {
            if(log.isTraceEnabled()) {
                log.trace(new StringBuffer("name of request correlator header (").append(hdr.corrName).
                          append(") is different from ours (").append(name).append("). Msg not accepted, passed up"));
            }
            return true;
        }

        // If the header contains a destination list, and we are not part of it, then we discard the
        // request (was addressed to other members)
        java.util.List dests=hdr.dest_mbrs;
        if(dests != null && local_addr != null && !dests.contains(local_addr)) {
            if(log.isTraceEnabled()) {
                log.trace(new StringBuffer("discarded request from ").append(msg.getSrc()).
                          append(" as we are not part of destination list (local_addr=").
                          append(local_addr).append(", hdr=").append(hdr).append(')'));
            }
            return false;
        }


        // [Header.REQ]:
        // i. If there is no request handler, discard
        // ii. Check whether priority: if synchronous and call stack contains
        // address that equals local address -> add priority request. Else
        // add normal request.
        //
        // [Header.RSP]:
        // Remove the msg request correlator header and notify the associated
        // <tt>RspCollector</tt> that a reply has been received
        switch(hdr.type) {
            case Header.REQ:
                if(request_handler == null) {
                    if(log.isWarnEnabled()) {
                        log.warn("there is no request handler installed to deliver request !");
                    }
                    return false;
                }

                if(deadlock_detection) {
                    if(scheduler == null) {
                        log.error("deadlock_detection is true, but scheduler is null: this is not supposed to happen" +
                                  " (discarding request)");
                        break;
                    }

                    Request req=new Request(msg);
                    java.util.Stack stack=hdr.callStack;
                    if(hdr.rsp_expected && stack != null && local_addr != null) {
                        if(stack.contains(local_addr)) {
                            if(log.isTraceEnabled())
                                log.trace("call stack=" + hdr.callStack + " contains " + local_addr +
                                          ": adding request to priority queue");
                            scheduler.addPrio(req);
                            break;
                        }
                    }
                    scheduler.add(req);
                    break;
                }

                handleRequest(msg);
                break;

            case Header.RSP:
                msg.removeHeader(name);
                RspCollector coll=findEntry(hdr.id);
                if(coll != null) {
                    Address sender=msg.getSrc();
                    Object retval=null;
                    byte[] buf=msg.getBuffer();
                    try {
                        retval=marshaller != null? marshaller.objectFromByteBuffer(buf) : Util.objectFromByteBuffer(buf);
                    }
                    catch(Exception e) {
                        log.error("failed unmarshalling buffer into return value", e);
                        try {
                            retval=marshaller != null? marshaller.objectToByteBuffer(e) : Util.objectToByteBuffer(e);
                        }
                        catch(Exception e1) {
                            log.error("failed marshalling exception " + e1 + " into buffer", e1);
                        }
                    }
                    coll.receiveResponse(retval, sender);
                }
                break;

            default:
                msg.removeHeader(name);
                if(log.isErrorEnabled()) log.error("header's type is neither REQ nor RSP !");
                break;
        }

        return false;
    }

    public Address getLocalAddress() {
        return local_addr;
    }

    public void setLocalAddress(Address local_addr) {
        this.local_addr=local_addr;
    }


    // .......................................................................

    /**
     * Add an association of:<br>
     * ID -> <tt>RspCollector</tt>
     */
    private void addEntry(long id, RequestEntry entry) {
        Long id_obj = new Long(id);
        synchronized(requests) {
            if(!requests.containsKey(id_obj))
                requests.put(id_obj, entry);
            else
                if(log.isWarnEnabled()) log.warn("entry " + entry + " for request-id=" + id + " already present !");
        }
    }


    /**
     * Remove the request entry associated with the given ID
     *
     * @param id the id of the <tt>RequestEntry</tt> to remove
     */
    private void removeEntry(long id) {
        Long id_obj = new Long(id);

        // changed by bela Feb 28 2003 (bug fix for 690606)
        // changed back to use synchronization by bela June 27 2003 (bug fix for #761804),
        // we can do this because we now copy for iteration (viewChange() and suspect())
        requests.remove(id_obj);
    }


    /**
     * @param id the ID of the corresponding <tt>RspCollector</tt>
     *
     * @return the <tt>RspCollector</tt> associated with the given ID
     */
    private RspCollector findEntry(long id) {
        Long id_obj = new Long(id);
        RequestEntry entry;

        entry=(RequestEntry)requests.get(id_obj);
        return((entry != null)? entry.coll:null);
    }


    /**
     * Handle a request msg for this correlator
     *
     * @param req the request msg
     */
    private void handleRequest(Message req) {
        Object        retval;
        byte[]        rsp_buf;
        Header        hdr, rsp_hdr;
        Message       rsp;

        // i. Remove the request correlator header from the msg and pass it to
        // the registered handler
        //
        // ii. If a reply is expected, pack the return value from the request
        // handler to a reply msg and send it back. The reply msg has the same
        // ID as the request and the name of the sender request correlator
        hdr=(Header)req.removeHeader(name);