mdg-filius/tools/java-oracle/src/TcpRun.java

// Runs a story of TCP calls on a core scenario's network in Java FILIUS
// (ScenarioNetwork) and records what really happens: every frame each card
// saw, when each call started, returned or failed, and every state each
// socket went through (polled every few milliseconds).
//
//   TcpRun <scenario.json> <story> <milliseconds> <out.json>
//
// The stories use the network of tcp-lan (server 192.168.0.10, clients
// 192.168.0.20 and .30) and run the calls as FILIUS' applications do, the
// server's in a thread of its own:
//   exchange    the client connects to port 80, sends "GET /", the server
//               answers with 3000 x (three segments); the client closes,
//               the server closes a second after it saw CLOSE_WAIT
//   closedPort  a client connects to port 81, where nothing listens
//   lostData    connected, all frames are lost (Verbindung.setDrop) while
//               the client sends; loss ends when senden fails
//   lostSynAck  the server's first SYN+ACK is lost
//   lostFin     connected, the client closes; its first FIN is lost
//   lostLastFin connected, the client closes; the server's first FIN is lost
//
// A single frame is lost on its sender's card (loseFirst): the card's output
// buffer drops it instead of queueing it. The packet log sees it on the
// sender, as with setDrop; unlike setDrop, the cable does not carry it.

import filius.hardware.Port;
import filius.hardware.Verbindung;
import filius.hardware.knoten.InternetKnoten;
import filius.software.netzzugangsschicht.EthernetFrame;
import filius.software.transportschicht.ServerSocket;
import filius.software.transportschicht.Socket;
import filius.software.transportschicht.TCPSocket;
import filius.software.transportschicht.TcpSegment;
import filius.software.vermittlungsschicht.IpPaket;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

public class TcpRun {
    static final String SERVER = "00000000-0000-0000-0000-000000000101";
    static final String CLIENT = "00000000-0000-0000-0000-000000000102";
    static final String OTHER_CLIENT = "00000000-0000-0000-0000-000000000103";

    static ScenarioNetwork network;
    static final List<Object[]> events = new ArrayList<>();
    static final Map<String, List<Object[]>> states = new LinkedHashMap<>();
    static final Map<String, TCPSocket> watched = new LinkedHashMap<>();

    static synchronized void event(String what) {
        events.add(new Object[] { System.currentTimeMillis() - network.start, what });
    }

    static synchronized void watch(String label, Socket socket) {
        watched.put(label, (TCPSocket) socket);
        states.put(label, new ArrayList<>());
        poll();
    }

    static synchronized void poll() {
        for (Map.Entry<String, TCPSocket> entry : watched.entrySet()) {
            List<Object[]> seen = states.get(entry.getKey());
            String state = entry.getValue().getStateAsString();
            if (seen.isEmpty() || !seen.get(seen.size() - 1)[1].equals(state)) {
                seen.add(new Object[] { System.currentTimeMillis() - network.start, state });
            }
        }
    }

    interface Call {
        Object run() throws Exception;
    }

    /** A call with its start and its end (result or exception) in the events. */
    static Object call(String label, Call call) {
        event(label + " called");
        try {
            Object result = call.run();
            event(label + " returned" + (result instanceof String ? " " + ((String) result).length() + " characters" : ""));
            return result;
        } catch (Exception e) {
            event(label + " failed " + e.getClass().getSimpleName());
            return null;
        }
    }

    static Thread server(Call body) {
        Thread thread = new Thread(() -> {
            try {
                body.run();
            } catch (Exception e) {
                event("server failed " + e.getClass().getSimpleName());
            }
        });
        thread.start();
        return thread;
    }

    /**
     * The node's card drops the first frame whose TCP segment matches, instead
     * of putting it into its output buffer. The cable waits on the buffer it
     * found: woken there once, it takes the new one.
     */
    @SuppressWarnings("unchecked")
    static void loseFirst(String nodeId, String label, Predicate<TcpSegment> which) throws Exception {
        Port port = ((InternetKnoten) network.system(nodeId).getKnoten()).getNetzwerkInterfaces().get(0).getPort();
        Field field = Port.class.getDeclaredField("ausgangsPuffer");
        field.setAccessible(true);
        LinkedList<EthernetFrame> old = (LinkedList<EthernetFrame>) field.get(port);
        field.set(port, new LinkedList<EthernetFrame>() {
            boolean lost;

            @Override
            public boolean add(EthernetFrame frame) {
                if (!lost && frame.getDaten() instanceof IpPaket
                        && ((IpPaket) frame.getDaten()).getSegment() instanceof TcpSegment
                        && which.test((TcpSegment) ((IpPaket) frame.getDaten()).getSegment())) {
                    lost = true;
                    event("lost " + label);
                    return true;
                }
                return super.add(frame);
            }
        });
        synchronized (old) {
            old.notifyAll();
        }
    }

    /** The listener's socket that waits for a connection, as soon as there is one. */
    static void watchListening(String label, ServerSocket listener) {
        Thread thread = new Thread(() -> {
            try {
                Field field = ServerSocket.class.getDeclaredField("aktuellerSocket");
                field.setAccessible(true);
                while (field.get(listener) == null) Thread.sleep(1);
                watch(label, (Socket) field.get(listener));
            } catch (Exception e) {
                event("watch failed " + e.getClass().getSimpleName());
            }
        });
        thread.setDaemon(true);
        thread.start();
    }

    /** Connected, then the client closes; the server's socket closes by itself. */
    static void connectAndClose() throws Exception {
        server(() -> {
            ServerSocket listener = new ServerSocket(network.system(SERVER), 80, IpPaket.TCP, new ArrayList<>());
            Socket accepted = (Socket) call("server oeffnen", listener::oeffnen);
            watch("server", accepted);
            return null;
        });
        Thread.sleep(500);
        TCPSocket client = new TCPSocket(network.system(CLIENT), "192.168.0.10", 80);
        watch("client", client);
        call("client verbinden", () -> {
            client.verbinden();
            return null;
        });
        Thread.sleep(500);
        call("client schliessen", () -> {
            client.schliessen();
            return null;
        });
    }

    static String repeat(String text, int count) {
        StringBuilder result = new StringBuilder();
        for (int i = 0; i < count; i++) result.append(text);
        return result.toString();
    }

    public static void main(String[] args) throws Exception {
        network = ScenarioNetwork.read(args[0]);
        String story = args[1];
        long duration = Long.parseLong(args[2]);
        switch (story) {
            case "lostSynAck":
                loseFirst(SERVER, "SYN+ACK", segment -> segment.isSyn() && segment.isAck());
                break;
            case "lostFin":
                loseFirst(CLIENT, "FIN", TcpSegment::isFin);
                break;
            case "lostLastFin":
                loseFirst(SERVER, "FIN", TcpSegment::isFin);
                break;
            default:
                break;
        }
        network.start();
        Thread poller = new Thread(() -> {
            while (true) {
                poll();
                try {
                    Thread.sleep(2);
                } catch (InterruptedException e) {
                    return;
                }
            }
        });
        poller.setDaemon(true);
        poller.start();
        long end = network.start + duration;

        switch (story) {
            case "exchange": {
                server(() -> {
                    ServerSocket listener = new ServerSocket(network.system(SERVER), 80, IpPaket.TCP, new ArrayList<>());
                    Socket accepted = (Socket) call("server oeffnen", listener::oeffnen);
                    watch("server", accepted);
                    call("server empfangen", accepted::empfangen);
                    call("server senden", () -> {
                        accepted.senden(repeat("x", 3000));
                        return null;
                    });
                    while (!((TCPSocket) accepted).getStateAsString().equals("CLOSE_WAIT")) Thread.sleep(2);
                    Thread.sleep(1000);
                    call("server schliessen", () -> {
                        accepted.schliessen();
                        return null;
                    });
                    return null;
                });
                Thread.sleep(500);
                TCPSocket client = new TCPSocket(network.system(CLIENT), "192.168.0.10", 80);
                watch("client", client);
                call("client verbinden", () -> {
                    client.verbinden();
                    return null;
                });
                call("client senden", () -> {
                    client.senden("GET /");
                    return null;
                });
                call("client empfangen", client::empfangen);
                call("client schliessen", () -> {
                    client.schliessen();
                    return null;
                });
                break;
            }
            case "closedPort": {
                Thread.sleep(500);
                TCPSocket client = new TCPSocket(network.system(OTHER_CLIENT), "192.168.0.10", 81);
                watch("client", client);
                call("client verbinden", () -> {
                    client.verbinden();
                    return null;
                });
                break;
            }
            case "lostData": {
                server(() -> {
                    ServerSocket listener = new ServerSocket(network.system(SERVER), 80, IpPaket.TCP, new ArrayList<>());
                    Socket accepted = (Socket) call("server oeffnen", listener::oeffnen);
                    watch("server", accepted);
                    return null;
                });
                Thread.sleep(500);
                TCPSocket client = new TCPSocket(network.system(CLIENT), "192.168.0.10", 80);
                watch("client", client);
                call("client verbinden", () -> {
                    client.verbinden();
                    return null;
                });
                Thread.sleep(500);
                event("loss on");
                Verbindung.setDrop(true);
                call("client senden", () -> {
                    client.senden("verloren");
                    return null;
                });
                Verbindung.setDrop(false);
                event("loss off");
                break;
            }
            case "lostSynAck": {
                server(() -> {
                    ServerSocket listener = new ServerSocket(network.system(SERVER), 80, IpPaket.TCP, new ArrayList<>());
                    watchListening("server", listener);
                    call("server oeffnen", listener::oeffnen);
                    return null;
                });
                Thread.sleep(500);
                TCPSocket client = new TCPSocket(network.system(CLIENT), "192.168.0.10", 80);
                watch("client", client);
                call("client verbinden", () -> {
                    client.verbinden();
                    return null;
                });
                break;
            }
            case "lostFin":
            case "lostLastFin":
                connectAndClose();
                break;
            default:
                throw new IllegalArgumentException("unknown story " + story);
        }

        long rest = end - System.currentTimeMillis();
        if (rest > 0) Thread.sleep(rest);
        poll();

        ScenarioNetwork.Json out = new ScenarioNetwork.Json();
        out.open('{');
        network.writeFrames(out);
        synchronized (TcpRun.class) {
            out.key("events").open('[');
            for (Object[] each : events) out.open('[').number((Long) each[0]).string((String) each[1]).close(']');
            out.close(']');
            out.key("states").open('{');
            for (Map.Entry<String, List<Object[]>> entry : states.entrySet()) {
                out.key(entry.getKey()).open('[');
                for (Object[] each : entry.getValue()) out.open('[').number((Long) each[0]).string((String) each[1]).close(']');
                out.close(']');
            }
            out.close('}');
        }
        out.close('}');
        network.stop();
        ScenarioNetwork.write(args[3], out);
        System.exit(0);
    }
}