tools/sci/org.eclipse.ptp.sci/libsci/routinglist.cpp - ptp/org.eclipse.ptp - Git at Google

 #ifndef _PRAGMA_COPYRIGHT_
 #define _PRAGMA_COPYRIGHT_
 #pragma comment(copyright, "%Z% %I% %W% %D% %T%\0")
 #endif /* _PRAGMA_COPYRIGHT_ */
 /****************************************************************************

 * Copyright (c) 2008, 2010 IBM Corporation.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0s
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html

  Classes: RoutingList

  Description: Provide routing services for all threads.

  Author: Nicole Nie, Liu Wei, Tu HongJ

  History:
    Date     Who ID    Description
    -------- --- ---   -----------
    05/08/09 nieyy        Initial code (D156654)
    01/16/12 ronglli      Add codes to retrieve BE list

 ****************************************************************************/

 #include <stdlib.h>
 #include <stdio.h>
 #include <assert.h>
 #include <string.h>

 #include <vector>

 #include "log.hpp"
 #include "packer.hpp"
 #include "group.hpp"
 #include "tools.hpp"
 #include "exception.hpp"
 #include "stream.hpp"
 #include "sshfunc.hpp"

 #include "routinglist.hpp"
 #include "eventntf.hpp"
 #include "initializer.hpp"
 #include "message.hpp"
 #include "queue.hpp"
 #include "ctrlblock.hpp"
 #include "filterproc.hpp"
 #include "readerproc.hpp"
 #include "topology.hpp"
 #include "writerproc.hpp"
 #include "eventntf.hpp"
 #include "dgroup.hpp"


 using namespace std;

 const int MAX_SUCCESSOR_NUM = 1024;
 const int MAX_SEGMENT_SIZE = 11680;
 const int MIN_SEGMENT_SIZE = 1440;  // 1500 - 40 - 20 (ethernet MTU - tcp/ip header - message header)

 RoutingList::RoutingList(int hndl)
     : handle(hndl), maxSegmentSize(MAX_SEGMENT_SIZE), filterProc(NULL), myDistriGroup(NULL), topology(NULL)
 {
     char *envp = ::getenv("SCI_SEGMENT_SIZE");
     if (envp != NULL) {
         maxSegmentSize = atoi(envp);
         maxSegmentSize = maxSegmentSize > MIN_SEGMENT_SIZE ? maxSegmentSize : MIN_SEGMENT_SIZE;
     }

     if (handle == -1) {
         // this is a front end, not parent
         myDistriGroup = new DistributedGroup(0);
     } else {
         int pid = -1;
         envp = ::getenv("SCI_PARENT_ID");
         if (envp) {
             pid = ::atoi(envp);
         } else {
             throw Exception(Exception::INVALID_LAUNCH);
         }
         myDistriGroup = new DistributedGroup(pid);
     }

     if (gCtrlBlock->getMyRole() != CtrlBlock::BACK_END) {
         topology = new Topology(0); // 0 is an impossible agent ID so it will be changed when it receives a real topology
     }
     successorList = new int[MAX_SUCCESSOR_NUM];
     queueInfo.clear();
     routers.clear();
     ::pthread_mutex_init(&mtx, NULL);
 }

 RoutingList::~RoutingList()
 {
     delete myDistriGroup;
     delete [] successorList;
     delete topology;
     ::pthread_mutex_destroy(&mtx);
 }

 void RoutingList::parseCmd(Message *msg)
 {
     bool notify = false;
     int rc = SCI_SUCCESS;
     if (msg->getType() == Message::GROUP_CREATE) {
         Packer packer(msg->getContentBuf());

         int num_bes = packer.unpackInt();
         int be_list[num_bes];
         for (int i=0; i<num_bes; i++) {
             be_list[i] = packer.unpackInt();
         }

         myDistriGroup->create(num_bes, be_list, msg->getGroup());
         bcast(msg->getGroup(), msg);

         if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
             notify = true;
         }
     } else if (msg->getType() == Message::GROUP_FREE) {
         sci_group_t group = msg->getGroup();

         bcast(group, msg);
         myDistriGroup->remove(group);

         if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
             notify = true;
         }
     } else if (msg->getType() == Message::GROUP_OPERATE) {
         Packer packer(msg->getContentBuf());

         sci_op_t op = (sci_op_t) packer.unpackInt();
         sci_group_t group1 = (sci_group_t) packer.unpackInt();
         sci_group_t group2 = (sci_group_t) packer.unpackInt();

         rc = myDistriGroup->operate(group1, group2, op, msg->getGroup());
         if (rc == SCI_SUCCESS) {
             bcast(msg->getGroup(), msg);
         }

         if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
             notify = true;
         }
     } else if (msg->getType() == Message::GROUP_OPERATE_EXT) {
         Packer packer(msg->getContentBuf());

         sci_op_t op = (sci_op_t) packer.unpackInt();
         sci_group_t group = (sci_group_t) packer.unpackInt();
         int num_bes = packer.unpackInt();
         int be_list[num_bes];
         for (int i=0; i<num_bes; i++) {
             be_list[i] = packer.unpackInt();
         }

         rc = myDistriGroup->operateExt(group, num_bes, be_list, op, msg->getGroup());
         if (rc == SCI_SUCCESS) {
             bcast(msg->getGroup(), msg);
         }

         if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
             notify = true;
         }
     } else if (msg->getType() == Message::GROUP_DROP) {
         myDistriGroup->dropSuccessor(msg->getID());
     } else if (msg->getType() == Message::GROUP_MERGE) {
         DistributedGroup subDistriGroup(-1);
         subDistriGroup.unpackMsg(*msg);

         if (subDistriGroup.getPID() == handle) {
             // if this message is from my son
             myDistriGroup->merge(msg->getID(), subDistriGroup, false);
         } else if (isSuccessorExist(subDistriGroup.getPID())){
             // if this message is from my grandson
             myDistriGroup->merge(msg->getID(), subDistriGroup, false);
         } else {
             // if this message is from my nephew
             myDistriGroup->merge(msg->getID(), subDistriGroup, true);

             // now update its parent id to me
             subDistriGroup.setPID(handle);

             // repack a message and send to my parent
             Message *newmsg = subDistriGroup.packMsg();
             filterProc->getOutQueue()->produce(newmsg);
         }
     } else {
         assert(!"should never be here");
     }

     if (notify) {
         void *ret = gNotifier->getRetVal(msg->getID());
         *((int *) ret) = rc;
         gNotifier->notify(msg->getID());
     }
 }

 void RoutingList::propagateGroupInfo()
 {
     // propgate my group information to my parent
     Message *msg = myDistriGroup->packMsg();
     if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
         filterProc->getOutQueue()->produce(msg);
     } else if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_END) {
         gCtrlBlock->getUpQueue()->produce(msg);
     } else {
         assert(!"should not be here");
     }
 }

 int RoutingList::getSegments(Message *msg, Message ***segments, int ref)
 {
     int i = 0;
     int segnum = (msg->getContentLen() + maxSegmentSize - 1) / maxSegmentSize + 1;
     int size = 0;
     char *ptr = msg->getContentBuf();
     sci_group_t gid = msg->getGroup();
     Message::Type typ = msg->getType();
     int mid = msg->getID();
     int mfid = msg->getFilterID();
     int hfid = mfid;
     int mlen = msg->getContentLen();
     *segments = (Message **)::malloc(segnum * sizeof(Message *));
     Message **segs = *segments;

     if ((mfid != SCI_FILTER_NULL) || (typ != Message::COMMAND)) {
         hfid = SCI_JOIN_SEGMENT;
     }
     ::memset(segs, 0, segnum * sizeof(Message *));
     segs[0] = new Message();
     segs[0]->build(hfid, gid, 0, NULL, NULL, Message::SEGMENT, segnum);
     segs[0]->setRefCount(ref);

     for (i = 1; i < segnum; i++) {
         segs[i] = new Message();
         size = (i < (segnum - 1)) ? maxSegmentSize : ((mlen - 1) % maxSegmentSize + 1);
         segs[i]->build(mfid, gid, 1, &ptr, &size, typ, mid);
         segs[i]->setRefCount(ref);
         ptr += size;
     }

     return segnum;
 }

 int RoutingList::bcast(sci_group_t group, Message *msg)
 {
     if (group > SCI_GROUP_ALL) {
         int hndl = querySuccessorId((int) group);
         if (hndl == INVLIDSUCCESSORID) {
             return SCI_ERR_GROUP_NOTFOUND;
         } else if (hndl == VALIDBACKENDIDS) {
             ucast((int)group, msg);
         } else {
             ucast(hndl, msg);
         }
         return SCI_SUCCESS;
     }

     if (!isGroupExist(group)) {
         return SCI_ERR_GROUP_NOTFOUND;
     }

     splitBcast(group, msg);

     return SCI_SUCCESS;
 }

 void RoutingList::splitBcast(sci_group_t group, Message *msg)
 {
     int numSor = numOfSuccessor(group);
     retrieveSuccessorList(group, successorList);
     mcast(msg, successorList, numSor);
 }

 void RoutingList::mcast(Message *msg, int *sorList, int num)
 {
     int i = 0;

     if (msg->getContentLen() <= maxSegmentSize) {
         msg->setRefCount(msg->getRefCount() + num);
         for (i = 0; i < num; i++) {
             queryQueue(sorList[i])->produce(msg);
         }
         return;
     }

     Message **segments;
     int segnum = getSegments(msg, &segments, num);
     for (i = 0; i < num; i++) {
         queryQueue(sorList[i])->multiProduce(segments, segnum);
     }
     ::free(segments);
 }

 void RoutingList::ucast(int successor_id, Message *msg, int refInc)
 {
     log_debug("Processor Router: send msg to successor %d", successor_id);
     mcast(msg, &successor_id, refInc);

     return;
 }

 void RoutingList::initSubGroup(int successor_id, int start_be_id, int end_be_id)
 {
     char qName[64] = {0};
     MessageQueue *queue = NULL;

     if (successor_id != VALIDBACKENDIDS) {
         queue = new MessageQueue();
         ::sprintf(qName, "Agent%d_inQ", successor_id);
         queue->setName(qName);
         mapQueue(successor_id, queue);
     } else {
         int i = 0;
         for (i = start_be_id; i <= end_be_id; i++) {
             queue = new MessageQueue();
             ::sprintf(qName, "BE%d_inQ", i);
             queue->setName(qName);
             mapQueue(i, queue);
         }
     }

     myDistriGroup->initSubGroup(successor_id, start_be_id, end_be_id);
 }

 int RoutingList::startReading(int hndl)
 {
     ROUTING_MAP::iterator it = routers.find(hndl);
     assert(it != routers.end());
     ReaderProcessor *reader = it->second.processor->getPeerProcessor();
     reader->start();

     return 0;
 }

 int RoutingList::startReaders()
 {
     ReaderProcessor *reader = NULL;
     ROUTING_MAP::iterator pit;

     for (pit = routers.begin(); pit != routers.end(); ++pit) {
         while (pit->second.processor == NULL) {
             SysUtil::sleep(WAIT_INTERVAL);
         }
         reader = pit->second.processor->getPeerProcessor();
         while (reader == NULL) {
             SysUtil::sleep(WAIT_INTERVAL);
             reader = pit->second.processor->getPeerProcessor();
         }
         reader->start();
     }

     return 0;
 }

 int RoutingList::numOfStreams()
 {
     int size;
     lock();
     size = routers.size();
     unlock();
     return size;
 }

 int RoutingList::getStreamsSockfds(int *fds)
 {
     int i = 0;
     ROUTING_MAP::iterator it;

     lock();
     for (it = routers.begin(); it != routers.end(); ++it) {
         fds[i] = it->second.stream->getSocket();
         i++;
     }
     unlock();

     return i;
 }

 int RoutingList::isActiveSockfd(int fd)
 {
     int isSocket = 0;
     ROUTING_MAP::iterator it;

     lock();
     for (it = routers.begin(); it != routers.end(); ++it) {
         if (fd == it->second.stream->getSocket()) {
             if ((it->second.stream->isReadActive()) || (it->second.stream->isWriteActive()) ){
                 isSocket = 1;
                 break;
             }
         }
     }
     unlock();

     return isSocket;
 }

 bool RoutingList::allActive()
 {
     bool active = true;
     ROUTING_MAP::iterator it;

     lock();
     for (it = routers.begin(); it != routers.end(); ++it) {
         if ((!(it->second.stream->isReadActive())) || (!(it->second.stream->isWriteActive())) ){
             active = false;
             break;
         }
     }
     unlock();

     return active;
 }

 void RoutingList::mapRouters(int hndl, WriterProcessor *writer, Stream *stream)
 {
     lock();
     if (writer != NULL) {
         routers[hndl].processor = writer;
     }
     routers[hndl].stream = stream;
     unlock();
 }

 int RoutingList::startRouting(int hndl, Stream *stream)
 {
     char name[64] = {0};
     MessageQueue *inQ = queryQueue(hndl);
     while (inQ == NULL) {
         SysUtil::sleep(WAIT_INTERVAL);
         inQ = queryQueue(hndl);
     }

     WriterProcessor *writer = NULL;
     if ((routers.find(hndl) != routers.end()) && (routers[hndl].stream != NULL)) {
         if (!gCtrlBlock->getRecoverMode()) {
             log_error("Duplicated client are trying to connect!!!");
             return -1;
         }
         writer = routers[hndl].processor;
         while(!(writer->getRecoverState())) {
             SysUtil::sleep(WAIT_INTERVAL);
         }

         writer->setOutStream(stream);
         mapRouters(hndl, NULL, stream);
         return 0;
     }

     log_debug("routers[%d].stream = %p", hndl, stream);
     ReaderProcessor *reader = new ReaderProcessor(hndl);
     reader->setInStream(stream);
     reader->setOutQueue(filterProc->getInQueue());
     ::sprintf(name, "Reader%d", hndl);
     reader->setName(name);

     writer = new WriterProcessor(hndl);
     writer->setInQueue(inQ);
     writer->setOutStream(stream);
     ::sprintf(name, "Writer%d", hndl);
     writer->setName(name);

     // reader is a peer processor of writer
     writer->setPeerProcessor(reader);
     reader->setPeerProcessor(writer);
     mapRouters(hndl, writer, stream);

     log_debug("The Reader%d thread has been newed!", hndl);
     writer->start();
     reader->start();

     return 0;
 }

 int RoutingList::stopRouting(int hndl)
 {
     ROUTING_MAP::iterator pit = routers.find(hndl);
     if (pit != routers.end()) {
         pit->second.processor->release();
         delete pit->second.processor;

         routers.erase(hndl);
         queueInfo.erase(hndl);
     }

     return 0;
 }

 int RoutingList::stopRouting()
 {
     // waiting for all processor threads terminate
     ROUTING_MAP::iterator pit;
     for (pit = routers.begin(); pit != routers.end(); ++pit) {
         pit->second.processor->release();
         delete pit->second.processor;
     }

     routers.clear();
     queueInfo.clear();

     return 0;
 }

 bool RoutingList::allRouted()
 {
     if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_AGENT) {
         char *envp = getenv("SCI_EMBED_AGENT");
         if ((envp != NULL) && (strcasecmp(envp, "yes") == 0)) {
             return (numOfQueues() == (numOfStreams() + 1));  // queueInfo contains itself
         }
     }

     return (numOfQueues() == numOfStreams());
 }

 void RoutingList::addBE(sci_group_t group, int successor_id, int be_id, bool init)
 {
     if (init) {
         char qName[64] = {0};
         int qID = 0;
         MessageQueue *queue = new MessageQueue();

         if (successor_id == VALIDBACKENDIDS) {
             qID = be_id;
             ::sprintf(qName, "BE%d_inQ", qID);
         } else {
             qID = successor_id;
             ::sprintf(qName, "Agent%d_inQ", qID);
         }
         queue->setName(qName);
         mapQueue(qID, queue);
     }

     myDistriGroup->addBE(group, successor_id, be_id);
 }

 void RoutingList::removeBE(int be_id)
 {
     myDistriGroup->removeBE(be_id);
 }

 void RoutingList::removeGroup(sci_group_t group)
 {
     myDistriGroup->remove(group);
 }

 void RoutingList::updateParentId(int pid)
 {
     ::setenv("SCI_PARENT_ID", SysUtil::itoa(pid).c_str(), 1);
     myDistriGroup->setPID(pid);
 }

 bool RoutingList::isGroupExist(sci_group_t group)
 {
     return myDistriGroup->isGroupExist(group);
 }

 bool RoutingList::isSuccessorExist(int successor_id)
 {
     return myDistriGroup->isSuccessorExist(successor_id);
 }


 int RoutingList::numOfBE(sci_group_t group)
 {
     return myDistriGroup->numOfBE(group);
 }

 int RoutingList::numOfSuccessor(sci_group_t group)
 {
     return myDistriGroup->numOfSuccessor(group);
 }

 int RoutingList::numOfBEOfSuccessor(int successor_id)
 {
     if (successor_id >= 0) {
         // if it is a back end
         return 1;
     }

     return myDistriGroup->numOfBEOfSuccessor(successor_id);
 }

 int RoutingList::querySuccessorId(int be_id)
 {
     return myDistriGroup->querySuccessorId(be_id);
 }

 void RoutingList::retrieveBEList(sci_group_t group, int * ret_val)
 {
     assert(ret_val);
     myDistriGroup->retrieveBEList(group, ret_val);
 }

 void RoutingList::retrieveSuccessorList(sci_group_t group, int * ret_val)
 {
     assert(ret_val);
     myDistriGroup->retrieveSuccessorList(group, ret_val);
 }

 void RoutingList::retrieveBEListOfSuccessor(int successor_id, int * ret_val)
 {
     assert(ret_val);
     myDistriGroup->retrieveBEListOfSuccessor(successor_id, ret_val);
 }

 void RoutingList::mapQueue(int hndl, MessageQueue *queue)
 {
     lock();
     queueInfo[hndl] = queue;
     unlock();
 }

 MessageQueue * RoutingList::queryQueue(int hndl)
 {
     MessageQueue *queue = NULL;

     lock();
     QUEUE_MAP::iterator qit = queueInfo.find(hndl);
     if (qit != queueInfo.end()) {
         queue = (*qit).second;
     }
     unlock();

     return queue;
 }

 int RoutingList::numOfQueues()
 {
     int size;
     lock();
     size = queueInfo.size();
     unlock();
     return size;
 }

 void RoutingList::lock()
 {
     ::pthread_mutex_lock(&mtx);
 }

 void RoutingList::unlock()
 {
     ::pthread_mutex_unlock(&mtx);
 }

 Topology * RoutingList::getTopology()
 {
     return topology;
 }

 FilterProcessor * RoutingList::getFilterProcessor()
 {
     return filterProc;
 }

 void RoutingList::setFilterProcessor(FilterProcessor *proc)
 {
     filterProc = proc;
 }
	#ifndef _PRAGMA_COPYRIGHT_
	#define _PRAGMA_COPYRIGHT_
	#pragma comment(copyright, "%Z% %I% %W% %D% %T%\0")
	#endif /* _PRAGMA_COPYRIGHT_ */
	/****************************************************************************

	* Copyright (c) 2008, 2010 IBM Corporation.
	* All rights reserved. This program and the accompanying materials
	* are made available under the terms of the Eclipse Public License v1.0s
	* which accompanies this distribution, and is available at
	* http://www.eclipse.org/legal/epl-v10.html

	Classes: RoutingList

	Description: Provide routing services for all threads.

	Author: Nicole Nie, Liu Wei, Tu HongJ

	History:
	Date Who ID Description
	-------- --- --- -----------
	05/08/09 nieyy Initial code (D156654)
	01/16/12 ronglli Add codes to retrieve BE list

	****************************************************************************/

	#include <stdlib.h>
	#include <stdio.h>
	#include <assert.h>
	#include <string.h>

	#include <vector>

	#include "log.hpp"
	#include "packer.hpp"
	#include "group.hpp"
	#include "tools.hpp"
	#include "exception.hpp"
	#include "stream.hpp"
	#include "sshfunc.hpp"

	#include "routinglist.hpp"
	#include "eventntf.hpp"
	#include "initializer.hpp"
	#include "message.hpp"
	#include "queue.hpp"
	#include "ctrlblock.hpp"
	#include "filterproc.hpp"
	#include "readerproc.hpp"
	#include "topology.hpp"
	#include "writerproc.hpp"
	#include "eventntf.hpp"
	#include "dgroup.hpp"


	using namespace std;

	const int MAX_SUCCESSOR_NUM = 1024;
	const int MAX_SEGMENT_SIZE = 11680;
	const int MIN_SEGMENT_SIZE = 1440; // 1500 - 40 - 20 (ethernet MTU - tcp/ip header - message header)

	RoutingList::RoutingList(int hndl)
	: handle(hndl), maxSegmentSize(MAX_SEGMENT_SIZE), filterProc(NULL), myDistriGroup(NULL), topology(NULL)
	{
	char *envp = ::getenv("SCI_SEGMENT_SIZE");
	if (envp != NULL) {
	maxSegmentSize = atoi(envp);
	maxSegmentSize = maxSegmentSize > MIN_SEGMENT_SIZE ? maxSegmentSize : MIN_SEGMENT_SIZE;
	}

	if (handle == -1) {
	// this is a front end, not parent
	myDistriGroup = new DistributedGroup(0);
	} else {
	int pid = -1;
	envp = ::getenv("SCI_PARENT_ID");
	if (envp) {
	pid = ::atoi(envp);
	} else {
	throw Exception(Exception::INVALID_LAUNCH);
	}
	myDistriGroup = new DistributedGroup(pid);
	}

	if (gCtrlBlock->getMyRole() != CtrlBlock::BACK_END) {
	topology = new Topology(0); // 0 is an impossible agent ID so it will be changed when it receives a real topology
	}
	successorList = new int[MAX_SUCCESSOR_NUM];
	queueInfo.clear();
	routers.clear();
	::pthread_mutex_init(&mtx, NULL);
	}

	RoutingList::~RoutingList()
	{
	delete myDistriGroup;
	delete [] successorList;
	delete topology;
	::pthread_mutex_destroy(&mtx);
	}

	void RoutingList::parseCmd(Message *msg)
	{
	bool notify = false;
	int rc = SCI_SUCCESS;
	if (msg->getType() == Message::GROUP_CREATE) {
	Packer packer(msg->getContentBuf());

	int num_bes = packer.unpackInt();
	int be_list[num_bes];
	for (int i=0; i<num_bes; i++) {
	be_list[i] = packer.unpackInt();
	}

	myDistriGroup->create(num_bes, be_list, msg->getGroup());
	bcast(msg->getGroup(), msg);

	if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
	notify = true;
	}
	} else if (msg->getType() == Message::GROUP_FREE) {
	sci_group_t group = msg->getGroup();

	bcast(group, msg);
	myDistriGroup->remove(group);

	if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
	notify = true;
	}
	} else if (msg->getType() == Message::GROUP_OPERATE) {
	Packer packer(msg->getContentBuf());

	sci_op_t op = (sci_op_t) packer.unpackInt();
	sci_group_t group1 = (sci_group_t) packer.unpackInt();
	sci_group_t group2 = (sci_group_t) packer.unpackInt();

	rc = myDistriGroup->operate(group1, group2, op, msg->getGroup());
	if (rc == SCI_SUCCESS) {
	bcast(msg->getGroup(), msg);
	}

	if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
	notify = true;
	}
	} else if (msg->getType() == Message::GROUP_OPERATE_EXT) {
	Packer packer(msg->getContentBuf());

	sci_op_t op = (sci_op_t) packer.unpackInt();
	sci_group_t group = (sci_group_t) packer.unpackInt();
	int num_bes = packer.unpackInt();
	int be_list[num_bes];
	for (int i=0; i<num_bes; i++) {
	be_list[i] = packer.unpackInt();
	}

	rc = myDistriGroup->operateExt(group, num_bes, be_list, op, msg->getGroup());
	if (rc == SCI_SUCCESS) {
	bcast(msg->getGroup(), msg);
	}

	if (gCtrlBlock->getMyRole() == CtrlBlock::FRONT_END) {
	notify = true;
	}
	} else if (msg->getType() == Message::GROUP_DROP) {
	myDistriGroup->dropSuccessor(msg->getID());
	} else if (msg->getType() == Message::GROUP_MERGE) {
	DistributedGroup subDistriGroup(-1);
	subDistriGroup.unpackMsg(*msg);

	if (subDistriGroup.getPID() == handle) {
	// if this message is from my son
	myDistriGroup->merge(msg->getID(), subDistriGroup, false);
	} else if (isSuccessorExist(subDistriGroup.getPID())){
	// if this message is from my grandson
	myDistriGroup->merge(msg->getID(), subDistriGroup, false);
	} else {
	// if this message is from my nephew
	myDistriGroup->merge(msg->getID(), subDistriGroup, true);

	// now update its parent id to me
	subDistriGroup.setPID(handle);

	// repack a message and send to my parent
	Message *newmsg = subDistriGroup.packMsg();
	filterProc->getOutQueue()->produce(newmsg);
	}
	} else {
	assert(!"should never be here");
	}

	if (notify) {
	void *ret = gNotifier->getRetVal(msg->getID());
	((int ) ret) = rc;
	gNotifier->notify(msg->getID());
	}
	}

	void RoutingList::propagateGroupInfo()
	{
	// propgate my group information to my parent
	Message *msg = myDistriGroup->packMsg();
	if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
	filterProc->getOutQueue()->produce(msg);
	} else if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_END) {
	gCtrlBlock->getUpQueue()->produce(msg);
	} else {
	assert(!"should not be here");
	}
	}

	int RoutingList::getSegments(Message msg, Message **segments, int ref)
	{
	int i = 0;
	int segnum = (msg->getContentLen() + maxSegmentSize - 1) / maxSegmentSize + 1;
	int size = 0;
	char *ptr = msg->getContentBuf();
	sci_group_t gid = msg->getGroup();
	Message::Type typ = msg->getType();
	int mid = msg->getID();
	int mfid = msg->getFilterID();
	int hfid = mfid;
	int mlen = msg->getContentLen();
	segments = (Message )::malloc(segnum sizeof(Message *));
	Message *segs = segments;

	if ((mfid != SCI_FILTER_NULL) \|\| (typ != Message::COMMAND)) {
	hfid = SCI_JOIN_SEGMENT;
	}
	::memset(segs, 0, segnum * sizeof(Message *));
	segs[0] = new Message();
	segs[0]->build(hfid, gid, 0, NULL, NULL, Message::SEGMENT, segnum);
	segs[0]->setRefCount(ref);

	for (i = 1; i < segnum; i++) {
	segs[i] = new Message();
	size = (i < (segnum - 1)) ? maxSegmentSize : ((mlen - 1) % maxSegmentSize + 1);
	segs[i]->build(mfid, gid, 1, &ptr, &size, typ, mid);
	segs[i]->setRefCount(ref);
	ptr += size;
	}

	return segnum;
	}

	int RoutingList::bcast(sci_group_t group, Message *msg)
	{
	if (group > SCI_GROUP_ALL) {
	int hndl = querySuccessorId((int) group);
	if (hndl == INVLIDSUCCESSORID) {
	return SCI_ERR_GROUP_NOTFOUND;
	} else if (hndl == VALIDBACKENDIDS) {
	ucast((int)group, msg);
	} else {
	ucast(hndl, msg);
	}
	return SCI_SUCCESS;
	}

	if (!isGroupExist(group)) {
	return SCI_ERR_GROUP_NOTFOUND;
	}

	splitBcast(group, msg);

	return SCI_SUCCESS;
	}

	void RoutingList::splitBcast(sci_group_t group, Message *msg)
	{
	int numSor = numOfSuccessor(group);
	retrieveSuccessorList(group, successorList);
	mcast(msg, successorList, numSor);
	}

	void RoutingList::mcast(Message msg, int sorList, int num)
	{
	int i = 0;

	if (msg->getContentLen() <= maxSegmentSize) {
	msg->setRefCount(msg->getRefCount() + num);
	for (i = 0; i < num; i++) {
	queryQueue(sorList[i])->produce(msg);
	}
	return;
	}

	Message **segments;
	int segnum = getSegments(msg, &segments, num);
	for (i = 0; i < num; i++) {
	queryQueue(sorList[i])->multiProduce(segments, segnum);
	}
	::free(segments);
	}

	void RoutingList::ucast(int successor_id, Message *msg, int refInc)
	{
	log_debug("Processor Router: send msg to successor %d", successor_id);
	mcast(msg, &successor_id, refInc);

	return;
	}

	void RoutingList::initSubGroup(int successor_id, int start_be_id, int end_be_id)
	{
	char qName[64] = {0};
	MessageQueue *queue = NULL;

	if (successor_id != VALIDBACKENDIDS) {
	queue = new MessageQueue();
	::sprintf(qName, "Agent%d_inQ", successor_id);
	queue->setName(qName);
	mapQueue(successor_id, queue);
	} else {
	int i = 0;
	for (i = start_be_id; i <= end_be_id; i++) {
	queue = new MessageQueue();
	::sprintf(qName, "BE%d_inQ", i);
	queue->setName(qName);
	mapQueue(i, queue);
	}
	}

	myDistriGroup->initSubGroup(successor_id, start_be_id, end_be_id);
	}

	int RoutingList::startReading(int hndl)
	{
	ROUTING_MAP::iterator it = routers.find(hndl);
	assert(it != routers.end());
	ReaderProcessor *reader = it->second.processor->getPeerProcessor();
	reader->start();

	return 0;
	}

	int RoutingList::startReaders()
	{
	ReaderProcessor *reader = NULL;
	ROUTING_MAP::iterator pit;

	for (pit = routers.begin(); pit != routers.end(); ++pit) {
	while (pit->second.processor == NULL) {
	SysUtil::sleep(WAIT_INTERVAL);
	}
	reader = pit->second.processor->getPeerProcessor();
	while (reader == NULL) {
	SysUtil::sleep(WAIT_INTERVAL);
	reader = pit->second.processor->getPeerProcessor();
	}
	reader->start();
	}

	return 0;
	}

	int RoutingList::numOfStreams()
	{
	int size;
	lock();
	size = routers.size();
	unlock();
	return size;
	}

	int RoutingList::getStreamsSockfds(int *fds)
	{
	int i = 0;
	ROUTING_MAP::iterator it;

	lock();
	for (it = routers.begin(); it != routers.end(); ++it) {
	fds[i] = it->second.stream->getSocket();
	i++;
	}
	unlock();

	return i;
	}

	int RoutingList::isActiveSockfd(int fd)
	{
	int isSocket = 0;
	ROUTING_MAP::iterator it;

	lock();
	for (it = routers.begin(); it != routers.end(); ++it) {
	if (fd == it->second.stream->getSocket()) {
	if ((it->second.stream->isReadActive()) \|\| (it->second.stream->isWriteActive()) ){
	isSocket = 1;
	break;
	}
	}
	}
	unlock();

	return isSocket;
	}

	bool RoutingList::allActive()
	{
	bool active = true;
	ROUTING_MAP::iterator it;

	lock();
	for (it = routers.begin(); it != routers.end(); ++it) {
	if ((!(it->second.stream->isReadActive())) \|\| (!(it->second.stream->isWriteActive())) ){
	active = false;
	break;
	}
	}
	unlock();

	return active;
	}

	void RoutingList::mapRouters(int hndl, WriterProcessor writer, Stream stream)
	{
	lock();
	if (writer != NULL) {
	routers[hndl].processor = writer;
	}
	routers[hndl].stream = stream;
	unlock();
	}

	int RoutingList::startRouting(int hndl, Stream *stream)
	{
	char name[64] = {0};
	MessageQueue *inQ = queryQueue(hndl);
	while (inQ == NULL) {
	SysUtil::sleep(WAIT_INTERVAL);
	inQ = queryQueue(hndl);
	}

	WriterProcessor *writer = NULL;
	if ((routers.find(hndl) != routers.end()) && (routers[hndl].stream != NULL)) {
	if (!gCtrlBlock->getRecoverMode()) {
	log_error("Duplicated client are trying to connect!!!");
	return -1;
	}
	writer = routers[hndl].processor;
	while(!(writer->getRecoverState())) {
	SysUtil::sleep(WAIT_INTERVAL);
	}

	writer->setOutStream(stream);
	mapRouters(hndl, NULL, stream);
	return 0;
	}

	log_debug("routers[%d].stream = %p", hndl, stream);
	ReaderProcessor *reader = new ReaderProcessor(hndl);
	reader->setInStream(stream);
	reader->setOutQueue(filterProc->getInQueue());
	::sprintf(name, "Reader%d", hndl);
	reader->setName(name);

	writer = new WriterProcessor(hndl);
	writer->setInQueue(inQ);
	writer->setOutStream(stream);
	::sprintf(name, "Writer%d", hndl);
	writer->setName(name);

	// reader is a peer processor of writer
	writer->setPeerProcessor(reader);
	reader->setPeerProcessor(writer);
	mapRouters(hndl, writer, stream);

	log_debug("The Reader%d thread has been newed!", hndl);
	writer->start();
	reader->start();

	return 0;
	}

	int RoutingList::stopRouting(int hndl)
	{
	ROUTING_MAP::iterator pit = routers.find(hndl);
	if (pit != routers.end()) {
	pit->second.processor->release();
	delete pit->second.processor;

	routers.erase(hndl);
	queueInfo.erase(hndl);
	}

	return 0;
	}

	int RoutingList::stopRouting()
	{
	// waiting for all processor threads terminate
	ROUTING_MAP::iterator pit;
	for (pit = routers.begin(); pit != routers.end(); ++pit) {
	pit->second.processor->release();
	delete pit->second.processor;
	}

	routers.clear();
	queueInfo.clear();

	return 0;
	}

	bool RoutingList::allRouted()
	{
	if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_AGENT) {
	char *envp = getenv("SCI_EMBED_AGENT");
	if ((envp != NULL) && (strcasecmp(envp, "yes") == 0)) {
	return (numOfQueues() == (numOfStreams() + 1)); // queueInfo contains itself
	}
	}

	return (numOfQueues() == numOfStreams());
	}

	void RoutingList::addBE(sci_group_t group, int successor_id, int be_id, bool init)
	{
	if (init) {
	char qName[64] = {0};
	int qID = 0;
	MessageQueue *queue = new MessageQueue();

	if (successor_id == VALIDBACKENDIDS) {
	qID = be_id;
	::sprintf(qName, "BE%d_inQ", qID);
	} else {
	qID = successor_id;
	::sprintf(qName, "Agent%d_inQ", qID);
	}
	queue->setName(qName);
	mapQueue(qID, queue);
	}

	myDistriGroup->addBE(group, successor_id, be_id);
	}

	void RoutingList::removeBE(int be_id)
	{
	myDistriGroup->removeBE(be_id);
	}

	void RoutingList::removeGroup(sci_group_t group)
	{
	myDistriGroup->remove(group);
	}

	void RoutingList::updateParentId(int pid)
	{
	::setenv("SCI_PARENT_ID", SysUtil::itoa(pid).c_str(), 1);
	myDistriGroup->setPID(pid);
	}

	bool RoutingList::isGroupExist(sci_group_t group)
	{
	return myDistriGroup->isGroupExist(group);
	}

	bool RoutingList::isSuccessorExist(int successor_id)
	{
	return myDistriGroup->isSuccessorExist(successor_id);
	}


	int RoutingList::numOfBE(sci_group_t group)
	{
	return myDistriGroup->numOfBE(group);
	}

	int RoutingList::numOfSuccessor(sci_group_t group)
	{
	return myDistriGroup->numOfSuccessor(group);
	}

	int RoutingList::numOfBEOfSuccessor(int successor_id)
	{
	if (successor_id >= 0) {
	// if it is a back end
	return 1;
	}

	return myDistriGroup->numOfBEOfSuccessor(successor_id);
	}

	int RoutingList::querySuccessorId(int be_id)
	{
	return myDistriGroup->querySuccessorId(be_id);
	}

	void RoutingList::retrieveBEList(sci_group_t group, int * ret_val)
	{
	assert(ret_val);
	myDistriGroup->retrieveBEList(group, ret_val);
	}

	void RoutingList::retrieveSuccessorList(sci_group_t group, int * ret_val)
	{
	assert(ret_val);
	myDistriGroup->retrieveSuccessorList(group, ret_val);
	}

	void RoutingList::retrieveBEListOfSuccessor(int successor_id, int * ret_val)
	{
	assert(ret_val);
	myDistriGroup->retrieveBEListOfSuccessor(successor_id, ret_val);
	}

	void RoutingList::mapQueue(int hndl, MessageQueue *queue)
	{
	lock();
	queueInfo[hndl] = queue;
	unlock();
	}

	MessageQueue * RoutingList::queryQueue(int hndl)
	{
	MessageQueue *queue = NULL;

	lock();
	QUEUE_MAP::iterator qit = queueInfo.find(hndl);
	if (qit != queueInfo.end()) {
	queue = (*qit).second;
	}
	unlock();

	return queue;
	}

	int RoutingList::numOfQueues()
	{
	int size;
	lock();
	size = queueInfo.size();
	unlock();
	return size;
	}

	void RoutingList::lock()
	{
	::pthread_mutex_lock(&mtx);
	}

	void RoutingList::unlock()
	{
	::pthread_mutex_unlock(&mtx);
	}

	Topology * RoutingList::getTopology()
	{
	return topology;
	}

	FilterProcessor * RoutingList::getFilterProcessor()
	{
	return filterProc;
	}

	void RoutingList::setFilterProcessor(FilterProcessor *proc)
	{
	filterProc = proc;
	}