<commit_before>package chabu.tester;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayDeque;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeMap;

import chabu.INetwork;
import chabu.INetworkUser;
import chabu.Utils;
import chabu.tester.data.ACommand;
import chabu.tester.data.AResult;
import chabu.tester.data.AXferItem;
import chabu.tester.data.CmdDutConnect;
import chabu.tester.data.CmdDutDisconnect;

public class ChabuTestNw {
	private String name;
	private boolean doShutDown;
	private Selector selector;
	
	private CtrlNetwork ctrlNw = new CtrlNetwork();
	private Thread thread;
	private boolean isStarted = false;

	private TreeMap<DutId, DutState> dutStates = new TreeMap<>();
	
	class DutState {
		
		final ByteBuffer txBuffer;
		final ArrayDeque<ACommand> commands;
		public SocketChannel socketChannel;
		int registeredInterrestOps = 0;
		public ByteBuffer rxBuffer;
		
		DutState(){
			commands = new ArrayDeque<>(100);
			txBuffer = ByteBuffer.allocate(2000);
			rxBuffer = ByteBuffer.allocate(2000);
			txBuffer.clear();
			rxBuffer.clear();
		}
		
		public void addCommand(ACommand cmd) throws ClosedChannelException {
			commands.add(cmd);
			if( (registeredInterrestOps & SelectionKey.OP_WRITE) == 0 ){
				selector.wakeup();
				registeredInterrestOps |= SelectionKey.OP_WRITE;
				socketChannel.register( selector, registeredInterrestOps, this );
			}
		}
		
	}
	
	class CtrlNetwork extends Network {
	
	}

	class Network implements INetwork {
		
//		ServerSocketChannel serverSocket;
//		SocketChannel socketChannel;
		ByteBuffer rxBuffer = ByteBuffer.allocate(0x10000);
		ByteBuffer txBuffer = ByteBuffer.allocate(0x10000);
		
		INetworkUser user;
		boolean userRequestRecv = false;
		boolean userRequestXmit = false;
		boolean netwRequestRecv = true;
		boolean netwRequestXmit = true;

		Network(){
			rxBuffer.position(rxBuffer.limit());
		}
		public void setNetworkUser(INetworkUser user) {
			this.user = user;
		}
		public void evUserXmitRequest() {
			userRequestXmit = true;
			netwRequestRecv = true;
			selector.wakeup();
		}
		public void evUserRecvRequest() {
			userRequestRecv = true;
			netwRequestXmit = true;
			selector.wakeup();
		}
		public void handleRequests() {
			if( userRequestRecv ){
				userRequestRecv = false;
				user.evRecv(rxBuffer);
			}
			if( userRequestXmit ){
				userRequestXmit = false;
				user.evXmit(txBuffer);
			}
		}
//		public void connectionClose() throws IOException {
//			socketChannel.close();
//		}
	};
	
	public void setCtrlNetworkUser(INetworkUser user) {
		this.ctrlNw.user = user;
	}
	
	public ChabuTestNw(String name) throws IOException, InterruptedException {

		this.name = name;
		
		selector = Selector.open();

//		ctrlNw.serverSocket = ServerSocketChannel.open();
//		ctrlNw.serverSocket.configureBlocking(false);
//		ctrlNw.serverSocket.register( selector, 0 );
		
		thread = new Thread( this::run, name );
		thread.start();
		synchronized(this){
			while( !isStarted ){
				this.wait();
			}
		}
	}
	
	private void run() {
		try {
			@SuppressWarnings("unused")
			int connectionOpenIndex = 0;

			synchronized(this){
				isStarted = true;
				notifyAll();
			}
			
			while (!doShutDown) {

				
//				System.out.printf("%s: selector sleep\n", name );
				selector.select();
//				System.out.printf("%s: selector wakeup\n", name );
				Set<SelectionKey> readyKeys = selector.selectedKeys();

				ctrlNw.handleRequests();
				Iterator<SelectionKey> iterator = readyKeys.iterator();
				while (iterator.hasNext()) {
					SelectionKey key = iterator.next();
					iterator.remove();
//					System.out.printf("%s selector key %s\n", name, key);
					if( key.isValid() ){
						if (key.isConnectable()) {
//							System.out.printf("%s: connecting\n", name );
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
								if (ds.socketChannel.isConnectionPending()){
									if( ds.socketChannel.finishConnect() ){
										ds.registeredInterrestOps &= ~SelectionKey.OP_CONNECT;
										ds.registeredInterrestOps |= SelectionKey.OP_READ;
										ds.socketChannel.register( selector, ds.registeredInterrestOps, ds );
										System.out.printf("%s: connecting ok\n", name );
									}
								}
							}
						}
						if (key.isWritable() ) {
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
//								System.out.printf("Tester %s: write p1\n", name );
								while( ds.txBuffer.remaining() > 1000 && !ds.commands.isEmpty() ){
									System.out.printf("Tester %s: loop buf %s\n", name, ds.txBuffer );
									ACommand cmd = ds.commands.remove();
									AXferItem.encodeItem(ds.txBuffer, cmd);
								}
//								System.out.printf("Tester %s: buf %s\n", name, ds.txBuffer );
								ds.txBuffer.flip();
//								System.out.printf("Tester %s: write %s bytes\n", name, ds.txBuffer.remaining() );
								ds.socketChannel.write(ds.txBuffer);
								if( !ds.txBuffer.hasRemaining() ){
									ds.registeredInterrestOps &= ~SelectionKey.OP_WRITE;
									ds.socketChannel.register( selector, ds.registeredInterrestOps, ds );
								}
								ds.txBuffer.compact();
							}
						}
						if( key.isReadable() ){
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
								int readSz = ds.socketChannel.read( ds.rxBuffer );
//								System.out.printf("%s read %d\n", name, readSz );
								ds.rxBuffer.flip();
								while( ds.rxBuffer.hasRemaining() ){
									AResult res = AResult.decodeResult(ds.rxBuffer);
									if( res == null ) {
										break;
									}
									consumeResult( res );
								}
								ds.rxBuffer.compact();
								if( readSz < 0 ){
									ds.socketChannel.close();
									System.out.printf("%s closed\n", name );
								}
							}
						}
					}<commit_msg>notify to unlock flush()<commit_after>package chabu.tester;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayDeque;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeMap;

import chabu.INetwork;
import chabu.INetworkUser;
import chabu.Utils;
import chabu.tester.data.ACommand;
import chabu.tester.data.AResult;
import chabu.tester.data.AXferItem;
import chabu.tester.data.CmdDutConnect;
import chabu.tester.data.CmdDutDisconnect;

public class ChabuTestNw {
	private String name;
	private boolean doShutDown;
	private Selector selector;
	
	private CtrlNetwork ctrlNw = new CtrlNetwork();
	private Thread thread;
	private boolean isStarted = false;

	private TreeMap<DutId, DutState> dutStates = new TreeMap<>();
	
	class DutState {
		
		final ByteBuffer txBuffer;
		final ArrayDeque<ACommand> commands;
		public SocketChannel socketChannel;
		int registeredInterrestOps = 0;
		public ByteBuffer rxBuffer;
		
		DutState(){
			commands = new ArrayDeque<>(100);
			txBuffer = ByteBuffer.allocate(2000);
			rxBuffer = ByteBuffer.allocate(2000);
			txBuffer.clear();
			rxBuffer.clear();
		}
		
		public void addCommand(ACommand cmd) throws ClosedChannelException {
			commands.add(cmd);
			if( (registeredInterrestOps & SelectionKey.OP_WRITE) == 0 ){
				selector.wakeup();
				registeredInterrestOps |= SelectionKey.OP_WRITE;
				socketChannel.register( selector, registeredInterrestOps, this );
			}
		}
		
	}
	
	class CtrlNetwork extends Network {
	
	}

	class Network implements INetwork {
		
//		ServerSocketChannel serverSocket;
//		SocketChannel socketChannel;
		ByteBuffer rxBuffer = ByteBuffer.allocate(0x10000);
		ByteBuffer txBuffer = ByteBuffer.allocate(0x10000);
		
		INetworkUser user;
		boolean userRequestRecv = false;
		boolean userRequestXmit = false;
		boolean netwRequestRecv = true;
		boolean netwRequestXmit = true;

		Network(){
			rxBuffer.position(rxBuffer.limit());
		}
		public void setNetworkUser(INetworkUser user) {
			this.user = user;
		}
		public void evUserXmitRequest() {
			userRequestXmit = true;
			netwRequestRecv = true;
			selector.wakeup();
		}
		public void evUserRecvRequest() {
			userRequestRecv = true;
			netwRequestXmit = true;
			selector.wakeup();
		}
		public void handleRequests() {
			if( userRequestRecv ){
				userRequestRecv = false;
				user.evRecv(rxBuffer);
			}
			if( userRequestXmit ){
				userRequestXmit = false;
				user.evXmit(txBuffer);
			}
		}
//		public void connectionClose() throws IOException {
//			socketChannel.close();
//		}
	};
	
	public void setCtrlNetworkUser(INetworkUser user) {
		this.ctrlNw.user = user;
	}
	
	public ChabuTestNw(String name) throws IOException, InterruptedException {

		this.name = name;
		
		selector = Selector.open();

//		ctrlNw.serverSocket = ServerSocketChannel.open();
//		ctrlNw.serverSocket.configureBlocking(false);
//		ctrlNw.serverSocket.register( selector, 0 );
		
		thread = new Thread( this::run, name );
		thread.start();
		synchronized(this){
			while( !isStarted ){
				this.wait();
			}
		}
	}
	
	private void run() {
		try {
			@SuppressWarnings("unused")
			int connectionOpenIndex = 0;

			synchronized(this){
				isStarted = true;
				notifyAll();
			}
			
			while (!doShutDown) {

				
//				System.out.printf("%s: selector sleep\n", name );
				selector.select();
//				System.out.printf("%s: selector wakeup\n", name );
				Set<SelectionKey> readyKeys = selector.selectedKeys();

				ctrlNw.handleRequests();
				Iterator<SelectionKey> iterator = readyKeys.iterator();
				while (iterator.hasNext()) {
					SelectionKey key = iterator.next();
					iterator.remove();
//					System.out.printf("%s selector key %s\n", name, key);
					if( key.isValid() ){
						if (key.isConnectable()) {
//							System.out.printf("%s: connecting\n", name );
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
								if (ds.socketChannel.isConnectionPending()){
									if( ds.socketChannel.finishConnect() ){
										ds.registeredInterrestOps &= ~SelectionKey.OP_CONNECT;
										ds.registeredInterrestOps |= SelectionKey.OP_READ;
										ds.socketChannel.register( selector, ds.registeredInterrestOps, ds );
										System.out.printf("%s: connecting ok\n", name );
									}
								}
							}
						}
						if (key.isWritable() ) {
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
//								System.out.printf("Tester %s: write p1\n", name );
								while( ds.txBuffer.remaining() > 1000 && !ds.commands.isEmpty() ){
									//System.out.printf("Tester %s: loop buf %s\n", name, ds.txBuffer );
									ACommand cmd = ds.commands.remove();
									AXferItem.encodeItem(ds.txBuffer, cmd);
								}
//								System.out.printf("Tester %s: buf %s\n", name, ds.txBuffer );
								ds.txBuffer.flip();
//								System.out.printf("Tester %s: write %s bytes\n", name, ds.txBuffer.remaining() );
								ds.socketChannel.write(ds.txBuffer);
								if( ds.commands.isEmpty() && !ds.txBuffer.hasRemaining() ){
									this.notifyAll();
								}
								if( !ds.txBuffer.hasRemaining() ){
									ds.registeredInterrestOps &= ~SelectionKey.OP_WRITE;
									ds.socketChannel.register( selector, ds.registeredInterrestOps, ds );
								}
								ds.txBuffer.compact();
								this.notifyAll();
							}
						}
						if( key.isReadable() ){
							synchronized( this ){
								DutState ds = (DutState)key.attachment();
								int readSz = ds.socketChannel.read( ds.rxBuffer );
//								System.out.printf("%s read %d\n", name, readSz );
								ds.rxBuffer.flip();
								while( ds.rxBuffer.hasRemaining() ){
									AResult res = AResult.decodeResult(ds.rxBuffer);
									if( res == null ) {
										break;
									}
									consumeResult( res );
								}
								ds.rxBuffer.compact();
								if( readSz < 0 ){
									ds.socketChannel.close();
									System.out.printf("%s closed\n", name );
								}
							}
						}
					}
{'fraction_non_alphanumeric': 0.12698803082472537, 'fraction_numerical': 0.0054107230693556324, 'mean_word_length': 3.0327272727272727, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7313417', 'n_tokens_mistral': 5136, 'n_tokens_neox': 3873, 'n_words': 768}