totalorder.java

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

// $Id: TotalOrder.java,v 1.13 2006/03/27 08:34:24 belaban Exp $


package org.jgroups.demos;

import org.jgroups.*;
import org.jgroups.util.Util;

import java.awt.*;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.awt.event.WindowAdapter;
import java.awt.event.WindowEvent;
import java.net.InetAddress;
import java.nio.ByteBuffer;


/**
 * Originally written to be a demo for TOTAL order (code to be written by a student). In the meantime,
 * it evolved into a state transfer demo. All members maintain a shared matrix and continually
 * broadcast changes to be applied to a randomly chosen field (e.g. multiplication of field with new
 * value, division, addition, subtraction). Each member can be started independently (starts to
 * broadcast update messages to all members). When "Stop" is pressed, a stop message is broadcast to
 * all members, causing them to stop sending messages. The "Clear" button clears the shared state;
 * "GetState" refreshes it from the shared group state (using the state transfer protocol).<p>If the
 * demo is to be used to show TOTAL order, then the TOTAL protocol would have to be added to the
 * stack.
 *
 * @author Bela Ban
 */
public class TotalOrder extends Frame {
    final Font def_font=new Font("Helvetica", Font.BOLD, 12);
    final Font def_font2=new Font("Helvetica", Font.PLAIN, 12);
    MyCanvas canvas;
    final MenuBar menubar=createMenuBar();
    final Button start=new Button("Start");
    final Button stop=new Button("Stop");
    final Button clear=new Button("Clear");
    final Button get_state=new Button("Get State");
    final Button quit=new Button("Quit");
    final Panel button_panel=new Panel();
    SenderThread sender=null;
    ReceiverThread receiver=null;
    Channel channel;
    Dialog error_dlg;
    long timeout=0;
    int field_size=0;
    int num_fields=0;
    static final int x_offset=30;
    static final int y_offset=40;
    private int num=0;

    private int num_additions=0, num_subtractions=0, num_divisions=0, num_multiplications=0;


    void error(String s) {
        System.err.println(s);
    }


    class EventHandler extends WindowAdapter {
        final Frame gui;

        public EventHandler(Frame g) {
            gui=g;
        }

        public void windowClosing(WindowEvent e) {
            gui.dispose();
            System.exit(0);
        }
    }


    class SenderThread extends Thread {
        TotOrderRequest req;
        boolean running=true;

        public void stopSender() {
            running=false;
            interrupt();
            System.out.println("-- num_additions: " + num_additions +
                    "\n-- num_subtractions: " + num_subtractions +
                    "\n-- num_divisions: " + num_divisions +
                    "\n-- num_multiplications: " + num_multiplications);
            num_additions=num_subtractions=num_multiplications=num_divisions=0;
        }

        public void run() {
            this.setName("SenderThread");

            byte[] buf;
            int cnt=0;
            while(running) {
                try {
                    req=createRandomRequest();
                    buf=req.toBuffer();
                    channel.send(new Message(null, null, buf));
                    System.out.print("-- num requests sent: " + cnt + "\r");
                    if(timeout > 0)
                        Util.sleep(timeout);
                    cnt++;
                    if(num > 0 && cnt > num) {
                        running=false;
                        cnt=0;
                    }
                }
                catch(Exception e) {
                    error(e.toString());
                    return;
                }
            }
        }
    }


    class ReceiverThread extends Thread {
        SetStateEvent set_state_evt;
        boolean running=true;


        public void stopReceiver() {
            running=false;
            interrupt();
        }

        public void run() {
            this.setName("ReceiverThread");
            Message msg;
            Object o;
            ByteBuffer buf;
            TotOrderRequest req;
            while(running) {
                try {
                    o=channel.receive(0);
                    if(o instanceof Message) {
                        try {
                            msg=(Message)o;
                            req=new TotOrderRequest();
                            buf=ByteBuffer.wrap(msg.getBuffer());
                            req.init(buf);
                            processRequest(req);
                        }
                        catch(Exception e) {
                            System.err.println(e);
                        }
                    }
                    else
                        if(o instanceof GetStateEvent) {
                            int[][] copy_of_state=canvas.getCopyOfState();
                            channel.returnState(Util.objectToByteBuffer(copy_of_state));
                        }
                        else
                            if(o instanceof SetStateEvent) {  // state was received, set it !
                                set_state_evt=(SetStateEvent)o;
                                canvas.setState(Util.objectFromByteBuffer(set_state_evt.getArg()));
                            }
                            else
                                if(o instanceof View) System.out.println(o.toString());
                }
                catch(ChannelClosedException closed) {
                    error("Channel has been closed; receiver thread quits");
                    return;
                }
                catch(Exception e) {
                    error(e.toString());
                    return;
                }
            }
        }
    }


    void processRequest(TotOrderRequest req) throws Exception {
        int x=req.x, y=req.y, val=req.val;

        if(req.type == TotOrderRequest.STOP) {
            stopSender();
            return;
        }

        switch(req.type) {
            case TotOrderRequest.ADDITION:
                canvas.addValueTo(x, y, val);
                num_additions++;
                break;
            case TotOrderRequest.SUBTRACTION:
                canvas.subtractValueFrom(x, y, val);
                num_subtractions++;
                break;
            case TotOrderRequest.MULTIPLICATION:
                canvas.multiplyValueWith(x, y, val);
                num_multiplications++;
                break;
            case TotOrderRequest.DIVISION:
                canvas.divideValueBy(x, y, val);
                num_divisions++;
                break;
        }
        canvas.update();
    }


    public TotalOrder(String title, long timeout, int num_fields, int field_size, String props, int num) {
        Dimension s;

        this.timeout=timeout;
        this.num_fields=num_fields;
        this.field_size=field_size;
        this.num=num;
        setFont(def_font);

        try {
            channel=new JChannel(props);
            channel.connect("TotalOrderGroup");
            channel.getState(null, 8000);
        }
        catch(Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }

        start.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                startSender();
            }
        });

        stop.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                try {
                    TotOrderRequest req=new TotOrderRequest(TotOrderRequest.STOP, 0, 0, 0);
                    byte[] buf=req.toBuffer();
                    channel.send(
                            new Message(
                                    null,
                                    null,
                                    buf));
                }
                catch(Exception ex) {
                }
            }
        });

        clear.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                canvas.clear();
            }
        });

        get_state.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                try {
                    boolean rc=channel.getState(null, 3000);
                    if(rc == false)
                        error("State could not be retrieved !");
                }
                catch(Throwable t) {
                    error("exception fetching state: " + t);
                }
            }
        });

        quit.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                channel.disconnect();
                channel.close();
                System.exit(0);
            }
        });

        setTitle(title);
        addWindowListener(new EventHandler(this));
        setBackground(Color.white);
        setMenuBar(menubar);

        setLayout(new BorderLayout());
        canvas=new MyCanvas(num_fields, field_size, x_offset, y_offset);

        add("Center", canvas);
        button_panel.setLayout(new FlowLayout());
        button_panel.setFont(def_font2);
        button_panel.add(start);
        button_panel.add(stop);
        button_panel.add(clear);
        button_panel.add(get_state);
        button_panel.add(quit);
        add("South", button_panel);

        s=canvas.getSize();
        s.height+=100;
        setSize(s);
        startReceiver();
    }


    void startSender() {
        if(sender == null || !sender.isAlive()) {
            sender=new SenderThread();
            sender.start();
        }
    }

    void stopSender() {
        if(sender != null) {
            sender.stopSender();
            sender=null;
        }
    }

    void startReceiver() {
        if(receiver == null) {
            receiver=new ReceiverThread();
            receiver.setPriority(Thread.MAX_PRIORITY);
            receiver.start();
        }
    }



    private MenuBar createMenuBar() {
        MenuBar ret=new MenuBar();
        Menu file=new Menu("File");
        MenuItem quitm=new MenuItem("Quit");

        ret.setFont(def_font2);
        ret.add(file);

        file.addSeparator();
        file.add(quitm);


        quitm.addActionListener(
                new ActionListener() {
                    public void actionPerformed(ActionEvent e) {
                        System.exit(1);
                    }
                });
        return ret;
    }



    private TotOrderRequest createRandomRequest() {
        TotOrderRequest ret=null;
        byte op_type=(byte)(((Math.random() * 10) % 4) + 1);  // 1 - 4
        int x=(int)((Math.random() * num_fields * 2) % num_fields);
        int y=(int)((Math.random() * num_fields * 2) % num_fields);
        int val=(int)((Math.random() * num_fields * 200) % 10);

        ret=new TotOrderRequest(op_type, x, y, val);
        return ret;
    }


    public static void main(String[] args) {
        TotalOrder g;
        String arg;
        long timeout=200;
        int num_fields=3;
        int field_size=80;
        String props=null;
        int num=0;

        props="UDP(mcast_addr=239.10.10.10;mcast_port=7500;ip_ttl=8;" +
                "mcast_send_buf_size=150000;mcast_recv_buf_size=80000):" +
                "PING(timeout=2000;num_initial_members=3):" +
                "MERGE2(min_interval=5000;max_interval=10000):" +
                "FD_SOCK:" +
                "VERIFY_SUSPECT(timeout=1500):" +
                "pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800):" +
                "UNICAST(timeout=5000):" +
                "pbcast.STABLE(desired_avg_gossip=20000):" +
                "FRAG(frag_size=4096;down_thread=false;up_thread=false):" +
                "pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;" +
                "shun=false;print_local_addr=true):" +
                "pbcast.STATE_TRANSFER";


        for(int i=0; i < args.length; i++) {
            arg=args[i];
            if("-timeout".equals(arg)) {
                timeout=Long.parseLong(args[++i]);
                continue;
            }
            if("-num_fields".equals(arg)) {