simplefac.java

SRI international 发布的OAA框架软件

package com.sri.oaa2.simplefac;

import com.sri.oaa2.icl.*;

import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;

import java.util.*;
import java.net.*;
import java.io.*;
import org.apache.log4j.Logger;
import org.apache.log4j.LogManager;
import org.apache.log4j.xml.DOMConfigurator;

/**
 * Simple version of an OAA2 facilitator.
 *
 * Handles agent registration.
 * Handles no reply and reply solvables.
 * Handles trace messages.
 */
public class SimpleFac implements Runnable
{
  // Logger
  static Logger logger = Logger.getLogger(SimpleFac.class.getName());
  static Logger loggerAddSolv = Logger.getLogger(SimpleFac.class.getName() + ".addSolvablesForId");

  // Handles incoming connections
  private ServerSocket listenStringSocket;

  // Handles incoming Java object connections
  private ServerSocket listenObjectSocket;

  // Handles incoming binary format connections
  private ServerSocket listenBinarySocket;

  // Maps id to a connection
  private HashMap idToConn = new HashMap();
  // Maps solvables to HashSet of ids that can solve those solvables
  private HashMap solvableToIds = new HashMap();
  // Maps id to HashSet of solvables that the address can solve
  private HashMap idToSolvables = new HashMap();

  // Next agent number to assign
  private static long nextAgentNum = 1;

  public SimpleFac() 
  {
    try {
      setServerStringSocket(new ServerSocket(3378));
      setServerBinarySocket(new ServerSocket(0));
    }
    catch(IOException ioe) {
      throw new RuntimeException("SimpleFac could not create listening socket(s): " + ioe.toString());
    }
  }

  public SimpleFac(int port) 
  {
    try {
      setServerStringSocket(new ServerSocket(port));
      setServerBinarySocket(new ServerSocket(0));
    }
    catch(IOException ioe) {
      throw new RuntimeException("SimpleFac could not create listening socket(s): " + ioe.toString());
    }
  }

  private final void setServerStringSocket(ServerSocket s) 
  {
    listenStringSocket = s;
  }
  
  protected final ServerSocket getServerStringSocket() 
  {
    return listenStringSocket;
  }

  private final void setServerBinarySocket(ServerSocket s) 
  {
    listenBinarySocket = s;
  }
  
  protected final ServerSocket getServerBinarySocket() 
  {
    return listenBinarySocket;
  }

  private final void setServerObjectSocket(ServerSocket s) 
  {
    listenObjectSocket = s;
  }
  
  protected final ServerSocket getServerObjectSocket() 
  {
    return listenObjectSocket;
  }

  public synchronized final long getNextAgentNum() 
  {
    long ret = nextAgentNum;
    ++nextAgentNum;
    return ret;
  }

  public final void run() 
  {
    SimpleFacListener stringListener;
    SimpleFacListener binaryListener;
    Semaphore readySem = new Semaphore(-2);
    stringListener = new SimpleFacListener(this, 
                                           getServerStringSocket(),
                                           new ConnectionFactory(FormatTypes.DEFAULT));
    binaryListener = new SimpleFacListener(this, 
                                           getServerBinarySocket(),
                                           new ConnectionFactory(FormatTypes.BINARY));
    stringListener.setInputReadySem(readySem);
    binaryListener.setInputReadySem(readySem);
    Thread stringThread = new Thread(stringListener);
    stringThread.setName("SimpleFacListener on listenStringSocket");
    Thread binaryThread = new Thread(binaryListener);
    binaryThread.setName("SimpleFacListener on listenBinarySocket");
    stringThread.start();
    binaryThread.start();
    try {
      readySem.acquire();
    }
    catch(InterruptedException ie) {
      logger.warn("SimpleFac.run() interrupted while waiting to acquire readySem");
    }
    if(logger.isInfoEnabled()) {
      logger.info("Ready.");
    }
    try {
      stringThread.join();
      binaryThread.join();
    }
    catch(InterruptedException ie) {
      return;
    }
  }

  /**
   * Release resources used by the agent with the given ID
   */
  protected final void releaseId(ConnectionId id) 
  {
    if(logger.isDebugEnabled()) {
      logger.debug("SimpleFac.releaseId() Releasing for " + id);
    }
    synchronized(this) {
      idToConn.remove(id);
      HashSet solvables = (HashSet)idToSolvables.remove(id);
      if(solvables == null) {
        return;
      }
      Iterator solvIt = solvables.iterator();
      HashSet solvToIdSet;
      Object key;
      while(solvIt.hasNext()) {
        key = solvIt.next();
        solvToIdSet = (HashSet)solvableToIds.get(key);
        if(solvToIdSet != null) {
          solvToIdSet.remove(id);
        }
        if(solvToIdSet.size() == 0) {
          solvableToIds.remove(key);
        }
      }
      if(logger.isDebugEnabled()) {
        logger.debug("SimpleFac.releaseId() solvableToIds: " + solvableToIds.toString());
        logger.debug("SimpleFac.releaseId() idToSolvables: " + idToSolvables.toString());
      }
    }
  }

  /**
   * Record solvables for the agent with teh given id
   */
  protected final void addSolvablesForId(ConnectionId id, HashSet solvables) 
  {
    synchronized(this) {
      idToSolvables.put(id, solvables);
      Iterator solvIt = solvables.iterator();
      Object key;
      while(solvIt.hasNext()) {
        key = solvIt.next();
        if(!solvableToIds.containsKey(key)) {
          if(logger.isDebugEnabled()) {
            logger.debug("Using new HashSet for key " + key);
          }
          HashSet s = new HashSet();
          s.add(id);
          solvableToIds.put(key, s);
        }
        HashSet ids = (HashSet)solvableToIds.get(key);
        ids.add(id);
      }
      if(loggerAddSolv.isDebugEnabled()) {
        loggerAddSolv.debug("SimpleFac.releaseId() solvableToIds: " + solvableToIds.toString());
        loggerAddSolv.debug("SimpleFac.releaseId() idToSolvables: " + idToSolvables.toString());
      }
    }
  }

  /**
   * Get solvables for a given ID
   */
  protected final HashSet getSolvablesForId(ConnectionId id) 
  {
    HashSet toRet;
    synchronized(this) {
      toRet = (HashSet)idToSolvables.get(id);
      if(toRet != null) {
        //toRet = (HashSet)toRet.clone();
        toRet = new HashSet(toRet);
      }
    }
    return toRet;
  }

  /**
   * Add a new connection
   */
  protected final void addConnection(ConnectionId id, SimpleFacConnection conn) 
  {
    synchronized(this) {
      idToConn.put(id, conn);
    }
  }

  /**
   * Construct OAA version 2 style listing of who can solve what
   */
  protected final IclTerm getV2AllSolvables() 
  {
    IclTerm toret;
    /*
    IclTerm readbb;
    IclTerm facSolvables;
    */
    synchronized(this) {
      // [read_bb(ksdata,[0,ready,[],root])]
      toret = IclTerm.fromString(true, "[read_bb(ksdata,[0,ready,[solvable(agent_data(_6862,_6863,_6864,_6865,_6866,_6867),[type(data)]),solvable(agent_host(_6848,_6849,_6850),[type(data)]),agent_version(_6834,_6835,_6836),solvable(facilitator_data(_6822,_6823,_6824,_6825,_6826),[type(data)]),can_solve(_6809,_6810),solvable(agent_location(_6798,_6799,_6800,_6801),[type(data)]),solvable(data(_6785,_6786),[type(data)])],root])]");
      /*
      toret = new IclList();
      readbb = new IclStruct("read_bb", new IclStr("ksdata"));
      facSolvables = new IclList(new IclInt(0), new IclStr("ready"), new IclList(), new IclStr("root"));
      readbb.add(facSolvables);
      toret.add(readbb);
      */
    }
    return toret;
  }
  
  /**
   * Get connection corresponding to an ID
   */
  protected final SimpleFacConnection getConnectionForId(ConnectionId id) 
  {
    SimpleFacConnection result;
    synchronized(this) {
      // note that any null return from get() is okay
      result = (SimpleFacConnection)idToConn.get(id);
    }
    return result;
  }

  /**
   * Replace all children in t with variables
   */
  private final IclTerm toStructAllVars(IclTerm t) 
  {
    IclStruct s = new IclStruct(t.toIdentifyingString());
    int varIndex = 0;
    int size = t.size();
    for(int i = 0; i < size; ++i) {
      StringBuffer b = new StringBuffer("_");
      b.append(Integer.toString(varIndex));
      s.add(new IclVar(b.toString()));
      ++varIndex;
    }
    return s;
  }

  /**
   * Get connections that can solve the given solvable.  This cheats and
   * assumes that all registered solvables are structs that contain only
   * variables.  It might also have to take into account the parameters.
   *
   * @param IclTerm solvable: solvable (standardized)
   * @return HashSet: copy of HashSet that lists ids (as Longs)
   */
  protected final HashSet whoCanSolve(IclTerm solvable) 
  {
    IclTerm toMatch = toStructAllVars(solvable);
    if(logger.isDebugEnabled()) {
      logger.debug("SimpleFac.whoCanSolve(): toMatch = [" + toMatch.toString() + "]");
    }
    HashSet ids = null;
    synchronized(this) {
      ids = (HashSet)solvableToIds.get(toMatch);
      if(ids == null) {
        return null;
      }
      ids = new HashSet(ids);
      //ids = (HashSet)ids.clone();
    }
    return ids;
  }

  /**
   * Get list of supported formats, for eg:
   *
   * [ascii(3378),default(3378),binary(3380)]
   */
  protected final IclTerm getSupportedFormatsList() 
  {
    IclInt asciiPort = new IclInt(getServerStringSocket().getLocalPort());
    IclInt defaultPort = new IclInt(getServerStringSocket().getLocalPort());
    IclInt binaryPort = new IclInt(getServerBinarySocket().getLocalPort());
    IclTerm ascii = new IclStruct("ascii", asciiPort);
    IclTerm def = new IclStruct("default", defaultPort);
    IclTerm binary = new IclStruct("binary", binaryPort);

    IclTerm toret;
    toret = new IclList(ascii, def, binary);
    //toret = new IclList(ascii, def);
    return toret;
  }

  public static void main(String[] args) throws Exception
  {
    String logconfigName = System.getProperty("OAALOGCONFIG", "com/sri/oaa2/simplefac/logconfig.xml");
    System.out.println("Using config file: [" + logconfigName + "]");
    if(ClassLoader.getSystemResourceAsStream(logconfigName) == null) {
      throw new RuntimeException("No such configuration file found: [" + logconfigName + "]");
    }
    if(logconfigName.equals("")) {
      throw new RuntimeException("No configuration file specified");
    }
    new DOMConfigurator().doConfigure(ClassLoader.getSystemResourceAsStream(logconfigName),
                                      LogManager.getLoggerRepository());
    SimpleFac fac = new SimpleFac();
    // Don't bother to create a new thread
    fac.run();
  }
}