322 lines
10 KiB
Java
322 lines
10 KiB
Java
/*
|
|
ReliableUDPSocket.java
|
|
*
|
|
* You may use and distribute under the terms of either the GNU Lesser
|
|
* General Public License, either version 2 of the license or,
|
|
* at your choice, any later version. Alternatively, you may use and
|
|
* distribute under the terms of the XPL.
|
|
*
|
|
* See the LICENSE.lgpl and LICENSE.xpl files for the specific terms of
|
|
* the licenses.
|
|
*
|
|
* This software is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the README
|
|
* file for more details.
|
|
*
|
|
*/
|
|
/* Written by Antti-Juhani Kaijanaho */
|
|
|
|
package org.gzigzag.net;
|
|
|
|
import org.gzigzag.*;
|
|
import java.io.*;
|
|
import java.net.*;
|
|
import java.util.*;
|
|
|
|
|
|
/** A socket for sending and receiving reliable UDP messages.
|
|
* This
|
|
* reliable UDP scheme guarantees delivery of a message but does not
|
|
* guarantee order of delivery.
|
|
* It does not specify a format for the
|
|
* UDP payload.
|
|
* Users of this class must implement
|
|
* ReliableUDPMessageFactory and pass that implementation to the
|
|
* constructor.
|
|
*/
|
|
public class ReliableUDPSocket implements SafeExit.Cleanupable {
|
|
public static final String rcsid = "$Id: ReliableUDPSocket.java,v 1.3 2001/03/18 20:22:11 tjl Exp $";
|
|
public static boolean dbg = false;
|
|
protected static void p(String s) { if(dbg) ZZLogger.log(s); }
|
|
protected static void pa(String s) { ZZLogger.log(s); }
|
|
|
|
/** Construct a reliable UDP socket and bind it to some local
|
|
* port. */
|
|
public ReliableUDPSocket(ReliableUDPMessageFactory fact) throws SocketException {
|
|
this(fact, new DatagramSocket());
|
|
}
|
|
|
|
/** Construct a reliable UDP socket and bind it to the given local
|
|
* port. */
|
|
public ReliableUDPSocket(ReliableUDPMessageFactory fact, int port) throws SocketException {
|
|
this(fact, new DatagramSocket(port));
|
|
}
|
|
|
|
private ReliableUDPSocket(ReliableUDPMessageFactory fact, DatagramSocket socket) {
|
|
this.fact = fact;
|
|
this.socket = socket;
|
|
this.local = new Peer(socket.getLocalAddress(), socket.getLocalPort());
|
|
qflusher.start();
|
|
}
|
|
|
|
/** Send the given message. */
|
|
public void send(ReliableUDPMessage rum) {
|
|
p("==> " + rum);
|
|
ReliableUDPMessage m = rum.copy();
|
|
m.setTimestamp(System.currentTimeMillis());
|
|
PeerEntry pe = getPeer(m);
|
|
pe.setseq(m);
|
|
pe.enqueue(m);
|
|
}
|
|
|
|
/** Receive a message. */
|
|
public ReliableUDPMessage receive() throws IOException {
|
|
synchronized (buf) {
|
|
DatagramPacket packet = new DatagramPacket(buf, buf.length);
|
|
ReliableUDPMessage m;
|
|
PeerEntry pe;
|
|
do {
|
|
socket.receive(packet);
|
|
byte[] bytes = new byte[packet.getLength()];
|
|
byte[] origbytes = packet.getData();
|
|
for (int i = 0; i < bytes.length; i++) bytes[i] = origbytes[i];
|
|
m = fact.makeMessage(bytes);
|
|
m.setPeer(new Peer(packet.getAddress(), packet.getPort()));
|
|
pe = getPeer(m);
|
|
pe.acknowledge(m);
|
|
pe.process_ack(m);
|
|
} while (pe.duplicate(m));
|
|
p("<== " + m);
|
|
return m;
|
|
}
|
|
}
|
|
|
|
/** Get the local address and port. */
|
|
public Peer getLocal() {
|
|
return local;
|
|
}
|
|
|
|
/** Get an array of currently active peers. */
|
|
public Peer[] getActivePeers() {
|
|
synchronized (peers) {
|
|
Peer[] rv = new Peer[peers.size()];
|
|
int i = 0;
|
|
for (Enumeration e = peers.elements(); e.hasMoreElements();) {
|
|
rv[i++] = (Peer)e.nextElement();
|
|
}
|
|
return rv;
|
|
}
|
|
}
|
|
|
|
/** Return a rough measure of the roundtrip time between us and
|
|
* the indicated peer, given in milliseconds. If the peer is not
|
|
* active, the return value is Long.MAX_VALUE. Note that this
|
|
* will not cause packets to be sent. */
|
|
public long getPeerLag(Peer p) {
|
|
synchronized (peers) {
|
|
if (!peers.containsKey(p)) return Long.MAX_VALUE;
|
|
PeerEntry pe = getPeer(p);
|
|
return pe.getMeasAverage();
|
|
}
|
|
}
|
|
|
|
public void close() {
|
|
try {
|
|
socket.close();
|
|
qflusher.interrupt();
|
|
qflusher.join();
|
|
qflusher = null;
|
|
} catch (InterruptedException e) {
|
|
Thread.currentThread().interrupt();
|
|
}
|
|
}
|
|
|
|
public void cleanup() {
|
|
close();
|
|
}
|
|
|
|
// PRIVATE STUFF
|
|
|
|
private final ReliableUDPMessageFactory fact;
|
|
private final DatagramSocket socket;
|
|
private final Peer local;
|
|
private final Hashtable peers = new Hashtable();
|
|
private final byte[] buf = new byte[65536];
|
|
private QueueFlusher qflusher = new QueueFlusher();
|
|
|
|
private final class PeerEntry {
|
|
private final class Outgoing {
|
|
public ReliableUDPMessage m;
|
|
public long time_entered = 0;
|
|
public long last_sent = 0;
|
|
}
|
|
|
|
private Vector outgoing = new Vector();
|
|
private Vector received = new Vector();
|
|
private boolean needAck = false;
|
|
private long ack = 0;
|
|
private long seq = 1;
|
|
private Peer peer;
|
|
private long lastSend = 0;
|
|
|
|
// for lag measurement
|
|
private long[] meas = new long[16]; // any number, really
|
|
private int measi = 0;
|
|
private int num_meas = 0;
|
|
|
|
public PeerEntry(Peer peer) { this.peer = peer; }
|
|
|
|
private synchronized void addReceived(Long l) {
|
|
received.addElement(l);
|
|
int n = received.size();
|
|
int i;
|
|
for (i = n - 1; i >= 1; i++) {
|
|
Long a = (Long)received.elementAt(i - 1);
|
|
Long b = (Long)received.elementAt(i);
|
|
if (a.longValue() < b.longValue()) break;
|
|
received.setElementAt(a, i);
|
|
received.setElementAt(b, i - 1);
|
|
}
|
|
if (n >= 2 && received.elementAt(i).equals(received.elementAt(i+1))) {
|
|
received.removeElementAt(i+1);
|
|
}
|
|
}
|
|
|
|
public synchronized boolean duplicate(ReliableUDPMessage rum) {
|
|
if (rum.getSeq() <= seq) return true;
|
|
if (received.contains(new Long(rum.getSeq()))) return true;
|
|
return false;
|
|
}
|
|
|
|
private synchronized void addMeasurement(long m) {
|
|
meas[measi] = m;
|
|
measi = (measi + 1) % meas.length;
|
|
if (num_meas < meas.length) ++num_meas;
|
|
}
|
|
|
|
public synchronized long getMeasAverage() {
|
|
if (num_meas == 0) return Long.MAX_VALUE;
|
|
long sum = 0;
|
|
for (int i = 0; i < num_meas; i++) {
|
|
sum += meas[i];
|
|
}
|
|
return sum / num_meas;
|
|
}
|
|
|
|
public synchronized void acknowledge(ReliableUDPMessage rum) {
|
|
needAck = true;
|
|
if (rum.getAck() > ack) addReceived(new Long(rum.getSeq()));
|
|
|
|
int n = received.size();
|
|
int i;
|
|
for (i = 0; i < n; i++) {
|
|
long l = ((Long)received.elementAt(i)).longValue();
|
|
if (l != ack + i + 1) break;
|
|
}
|
|
Vector v = new Vector();
|
|
ack += i;
|
|
for (int j = i; j < n; i++) {
|
|
v.addElement(received.elementAt(i));
|
|
}
|
|
received = v;
|
|
}
|
|
|
|
public synchronized void process_ack(ReliableUDPMessage rum) {
|
|
int n = outgoing.size();
|
|
Vector v = new Vector();
|
|
for (int i = 0; i < n; i++) {
|
|
Outgoing og = (Outgoing)outgoing.elementAt(i);
|
|
if (og.m.getSeq() > rum.getAck()) {
|
|
v.addElement(og);
|
|
} else if (og.m.getSeq() == rum.getAck()) {
|
|
addMeasurement(System.currentTimeMillis() - og.last_sent);
|
|
}
|
|
}
|
|
outgoing = v;
|
|
}
|
|
|
|
public synchronized void enqueue(ReliableUDPMessage m) {
|
|
Outgoing og = new Outgoing();
|
|
og.m = m;
|
|
og.time_entered = System.currentTimeMillis();
|
|
og.last_sent = 0;
|
|
|
|
outgoing.addElement(og);
|
|
trySend(og);
|
|
}
|
|
|
|
public synchronized void setseq(ReliableUDPMessage rum) {
|
|
rum.setSeq(seq++);
|
|
}
|
|
|
|
public synchronized void flush() {
|
|
int n = outgoing.size();
|
|
for (int i = 0; i < n; i++) {
|
|
Outgoing og = (Outgoing)outgoing.elementAt(i);
|
|
trySend(og);
|
|
if (og.last_sent - og.time_entered > 10 * 60 * 1000) {
|
|
peers.remove(peer);
|
|
return;
|
|
}
|
|
}
|
|
if (needAck) {
|
|
Outgoing og = new Outgoing();
|
|
og.m = fact.makeNoOp();
|
|
og.m.setSeq(0);
|
|
trySend(og);
|
|
}
|
|
}
|
|
|
|
private void trySend(Outgoing og) {
|
|
long now = System.currentTimeMillis();
|
|
if (now - og.last_sent < 500) return;
|
|
og.m.setAck(ack);
|
|
needAck = false;
|
|
lastSend = now;
|
|
byte[] bytes = og.m.toByteArray();
|
|
DatagramPacket dp = new DatagramPacket(bytes, bytes.length,
|
|
peer.getHost(), peer.getPort());
|
|
try {
|
|
socket.send(dp);
|
|
} catch (IOException e) {
|
|
needAck = true;
|
|
}
|
|
og.last_sent = now;
|
|
}
|
|
|
|
}
|
|
|
|
private PeerEntry getPeer(ReliableUDPMessage rum) {
|
|
Peer p = rum.getPeer();
|
|
return getPeer(p);
|
|
}
|
|
|
|
private PeerEntry getPeer(Peer p) {
|
|
synchronized (peers) {
|
|
if (!peers.containsKey(p)) {
|
|
peers.put(p, new PeerEntry(p));
|
|
}
|
|
return (PeerEntry)peers.get(p);
|
|
}
|
|
}
|
|
|
|
private final class QueueFlusher extends Thread {
|
|
public void run() {
|
|
while (!isInterrupted()) {
|
|
try {
|
|
sleep(500);
|
|
Peer[] ps = getActivePeers();
|
|
for (int i = 0; i < ps.length; i++) {
|
|
getPeer(ps[i]).flush();
|
|
yield();
|
|
}
|
|
} catch (InterruptedException e) {
|
|
interrupt();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|