Java APIs

I have a code that is trying to create separate threads for each Node in a Graph<Node, Edge>. Within the thread run() method, each Node creates as many SocketChannel-s as its number of neighbor Node-s and try to read or write with its neighbors via the SelectionKey class. Note: I am using java.nio.* package to handle non-blocking (asynchronous) communication. Another reason for its usage is because a single selector can handle multiple SocketChannels, so the multiplexing overhead will be low.

Below is how my code goes:

public class Node extends Object implements Runnable, Serializable {
//public static variables
//private static variables
//private variables

public Node() throws UnknownHostException {
this.nid = IDseq++ ;
Random gen = new Random(System.nanoTime());
this.setValue(gen.nextInt(100));
hostname = InetAddress.getLocalHost().getHostName();
socketAddress = new InetSocketAddress(hostname, ++port);
System.out.println("Node started at : " + this.socketAddress);
}
public void run() {
System.out.println("Node " + this.toString() + ": Thread running ...");
//add as many SocketChannels to the SocketChannelPool as the number of node neighbors
sChannelPool = new SocketChannelPool(nMap.size(),socketAddress.getHostName(),socketAddress.getPort());
for (Node nbh : nMap.keySet())
sChannelPool.addChannel();
//register the SocketChannels with the selector
try {
selector = Selector.open();
for (SocketChannel sc : sChannelPool.backingList) {
sc.register(selector, sc.validOps());
}
} catch (IOException exception) {
System.out.println(exception.getMessage());
System.exit(1);
}

//wait for reading from or writing to SocketChannels
while (true && iterations < MAX_ITER) {
try {
int readyKeys = selector.select();
if (readyKeys <= 0)
continue;
} catch (IOException exception) {
System.out.println(exception.getMessage());
}

Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
while (selectedKeys.hasNext()) {
SelectionKey selectedKey = selectedKeys.next();
selectedKeys.remove();
if (!selectedKey.isValid())
continue;
try {
if (selectedKey.isAcceptable()) {
acceptKey(selectedKey);
} else if (selectedKey.isReadable()) {
double prevVal = value;
readKey(selectedKey);
if (Math.abs(prevVal - value) < STOP_THRESHOLD) {
break;
}
iterations++;
} else if (selectedKey.isWritable()) {
writeKey(selectedKey);
}
} catch (CancelledKeyException exception) {
System.out.println(exception.getMessage());
closeKey(selectedKey);
} catch (ClassNotFoundException exception) {
System.out.println(exception.getMessage());
closeKey(selectedKey);
} catch (IOException exception) {
System.out.println(exception.getMessage());
closeKey(selectedKey);
}
}
} //while (true)

System.out.println("Node " + this.toString() + ": Thread stopped");
}
}

My Main code consists of the following code

public class Main {
public static void main(String[] args) throws InterruptedException {
//generate graph, add vertices and edges

//start the communication between the nodes
for (Node n : graph.vertexSet()) {
(new Thread(n)).start();
}
}

}

Now, when I am running this code from Main with runtime arguments of 9 Nodes and 12 Edges, it is giving me the following error :

Node nodes.Node@59e152c5: Thread running ...
Node nodes.Node@5801319c: Thread running ...
Node nodes.Node@4cbfea1d: Thread running ...
Node nodes.Node@65b4fad5: Thread running ...
Node nodes.Node@522a4983: Thread running ...
Node nodes.Node@5a77a7f9: Thread running ...
Node nodes.Node@366025e7: Thread running ...
Node nodes.Node@1cc7b00c: Thread running ...
Node nodes.Node@56406199: Thread running ...
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source)
at sun.nio.ch.SocketAdaptor.connect(Unknown Source)
at io.SocketChannelPool.addChannel(SocketChannelPool.java:47)
at nodes.Node.run(Node.java:100)
at java.lang.Thread.run(Unknown Source)
...


Update

The exception is given in line 47 of SocketChannelPool.java (the bold line - socketChannel.socket().connect(endPoint, MessageConstants.CONNECT_TIMEOUT) ) as below:

public class SocketChannelPool {
//backing store for the pool
public LinkedList<SocketChannel> backingList = new LinkedList<SocketChannel>();
//maximum size of the pool
private int size;
//host to which new channels should be connected
private String host;
//port to which new channels should be connected
private int port;
//constructor
public SocketChannelPool(int size, String host, int port) {
this.size = size;
this.host = host;
this.port = port;
}
//add channels to the pool
public void addChannel() {
//new channel is added to front of backing list
SocketChannel socketChannel = null;
InetSocketAddress endPoint = new InetSocketAddress(host, port);
try {
socketChannel = SocketChannel.open();
socketChannel.socket().setReceiveBufferSize(MessageConstants.SOCKET_BUFFER_SIZE);
socketChannel.socket().setSendBufferSize(MessageConstants.SOCKET_BUFFER_SIZE);
socketChannel.socket().connect(endPoint, MessageConstants.CONNECT_TIMEOUT);
socketChannel.configureBlocking(false);
} catch (IOException e1) {
e1.printStackTrace();
}
backingList.addFirst(socketChannel);
//if the size of backing list is exceeded, grab the last entry and close it
if (backingList.size() > size) {
try {
backingList.removeLast().close();
} catch (IOException e) {
//Logs the message and stops the current operation
System.err.println("Failed to close socket channel when pool size exceeded: " + e.getMessage());
System.exit(1);
}
}
}







The error java.net.ConnectException is repeated for each of the Nodes.

I am confused about whether this error has anything to do with synchronization problem or is it due to some SocketChannel issue. From the description of the error, it seems that it can't connect to the host and port in the line 47: SocketChannel.socket().connect(new InetSocketAddress(host, port), timeout). I am running all the 9 Nodes on the same localhost machine but different ports. Is that's the reason for this error. Any help will be highly appreciated

Thanks
Somnath

Edited by: user11269950 on Sep 30, 2012 6:04 AM