Apocrita: A Distributed Peer-to-Peer File Sharing System  for Intranets  Joshua J. Reynolds, Robbie McLeod, Qusay H. Mahmoud  Distributed Computing and Wireless & Telecommunications Technology  University of Guelph-Humber  Toronto, ON, M9W 5L7 Canada  {jreyno04,rmcleo01,qmahmoud}@uoguelph.ca  ABSTRACT  Many organizations are required to author documents for various  purposes, and such documents may need to be accessible by all  member of the organization. This access may be needed for  editing or simply viewing a document. In some cases these  documents are shared between authors, via email, to be edited.  This can easily cause incorrect version to be sent or conflicts  created between multiple users trying to make amendments to a  document. There may even be multiple different documents in the  process of being edited. The user may be required to search for a  particular document, which some search tools such as Google  Desktop may be a solution for local documents but will not find a  document on another user"s machine. Another problem arises  when a document is made available on a user"s machine and that  user is offline, in which case the document is no longer  accessible. In this paper we present Apocrita, a revolutionary  distributed P2P file sharing system for Intranets.  Categories and Subject Descriptors  C.2.4 [Computer-Communication Networks]: Distributed  Systems - Distributed applications.    1. INTRODUCTION  The Peer-to-Peer (P2P) computing paradigm is becoming a  completely new form of mutual resource sharing over the  Internet. With the increasingly common place broadband Internet  access, P2P technology has finally become a viable way to share  documents and media files.  There are already programs on the market that enable P2P file  sharing. These programs enable millions of users to share files  among themselves. While the utilization of P2P clients is already  a gigantic step forward compared to downloading files off  websites, using such programs are not without their problems.  The downloaded files still require a lot of manual management by  the user. The user still needs to put the files in the proper  directory, manage files with multiple versions, delete the files  when they are no longer wanted. We strive to make the process of  sharing documents within an Intranet easier.  Many organizations are required to author documents for various  purposes, and such documents may need to be accessible by all  members of the organization. This access may be needed for  editing or simply viewing a document. In some cases these  documents are sent between authors, via email, to be edited. This  can easily cause incorrect version to be sent or conflicts created  between multiple users trying to make amendments to a  document. There may even be multiple different documents in the  process of being edited. The user may be required to search for a  particular document, which some search tools such as Google  Desktop may be a solution for local documents but will not find a  document on another user"s machine. Furthermore, some  organizations do not have a file sharing server or the necessary  network infrastructure to enable one. In this paper we present  Apocrita, which is a cost-effective distributed P2P file sharing  system for such organizations.  The rest of this paper is organized as follows. In section 2, we  present Apocrita. The distributed indexing mechanism and  protocol are presented in Section 3. Section 4 presents the   peer-topeer distribution model. A proof of concept prototype is presented  in Section 5, and performance evaluations are discussed in  Section 6. Related work is presented is Section 7, and finally  conclusions and future work are discussed in Section 8.  2. APOCRITA  Apocrita is a distributed peer-to-peer file sharing system, and has  been designed to make finding documents easier in an Intranet  environment. Currently, it is possible for documents to be located  on a user's machine or on a remote machine. It is even possible  that different revisions could reside on each node on the Intranet.  This means there must be a manual process to maintain document  versions. Apocrita solves this problem using two approaches.  First, due to the inherent nature of Apocrita, the document will  only reside on a single logical location. Second, Apocrita provides  a method of reverting to previous document versions. Apocrita  Permission to make digital or hard copies of all or part of this work for  personal or classroom use is granted without fee provided that copies are  not made or distributed for profit or commercial advantage and that  copies bear this notice and the full citation on the first page. To copy  otherwise, or republish, to post on servers or to redistribute to lists,  requires prior specific permission and/or a fee.  ACMSE"07, MARCH 23-24, 2007, WINSTON-SALEM, NC, USA.  COPYRIGHT 2007 ACM 978-1-59593-629-5/07/0003 …$5.00.  174  will also distribute documents across multiple machines to ensure  high availability of important documents. For example, if a  machine contains an important document and the machine is  currently inaccessible, the system is capable of maintaining  availability of the document through this distribution mechanism.  It provides a simple interface for searching and accessing files  that may exist either locally or remotely. The distributed nature of  the documents is transparent to the user. Apocrita supports a  decentralized network model where the peers use a discovery  protocol to determine peers.  Apocrita is intended for network users on an Intranet. The main  focus is organizations that may not have a network large enough  to require a file server and supporting infrastructure. It eliminates  the need for documents to be manually shared between users  while being edited and reduces the possibility of conflicting  versions being distributed. The system also provides some  redundancy and in the event of a single machine failure, no  important documents will be lost. It is operating system  independent, and easy to access through a web browser or through  a standalone application. To decrease the time required for  indexing a large number of documents, the indexing process is  distributed across available idle nodes. Local and remote files  should be easily accessible through a virtual mountable file  system, providing transparency for users.  3. DISTRIBUTED INDEXING  Apocrita uses a distributed index for all the documents that are  available on the Intranet. Each node will contain part of the full  index, and be aware of what part of the index each other node has.  A node will be able to contact each node that contains a unique  portion of the index. In addition, each node has a separate local  index of its own documents. But as discussed later, in the current  implementation, each node has a copy of the entire index.  Indexing of the documents is distributed. Therefore, if a node is in  the process of indexing many documents, it will break up the  work over the nodes. Once a node"s local index is updated with  the new documents, the distributed index will then be updated.  The current distributed indexing system consists of three separate  modules: NodeController, FileSender, and NodeIndexer. The  responsibility of each module is discussed later in this section.  3.1 Indexing Protocol  The protocol we have designed for the distributed indexing is  depicted in Figure 1.  Figure 1. Apocrita distributed indexing protocol.  IDLE QUERY: The IDLE QUERY is sent out from the initiating  node to determine which other nodes may be able to help with the  overall indexing process. There are no parameters sent with the  command. The receiving node will respond with either a BUSY  or IDLE command. If the IDLE command is received, the  initiating node will add the responding node to a list of available  distributed indexing helpers. In the case of a BUSY command  being received, the responding node is ignored.  BUSY: Once a node received an IDL QUERY, it will determine  whether it can be considered a candidate for distributed indexing.  This determination is based on the overall CPU usage of the node.  If the node is using most of its CPU for other processes, the node  will respond to the IDLE QUERY with a BUSY command.  IDLE: As with the case of the BUSY response, the node  receiving the IDLE QUERY will determine its eligibility for  distributed indexing. To be considered a candidate for distributed  indexing, the overall CPU usage must be at a minimum to all for  dedicated indexing of the distributed documents. If this is the  case, the node will respond with an IDLE command.  INCOMING FILE: Once the initiating node assembles a set of  idle nodes to assist with the distributed indexing, it will divide the  documents to be sent to the nodes. To do this, it sends an  INCOMING FILE message, which contains the name of the file  as well as the size in bytes. After the INCOMING FILE command  has been sent, the initiating node will begin to stream the file to  the other node. The initiating node will loop through the files that  are to be sent to the other node; each file stream being preceded  by the INCOMING FILE command with the appropriate  parameters.  INDEX FILE: Once the indexing node has completed the  indexing process of the set of files, it must send the resultant  index back to the initiating node. The index is comprised of  multiple files, which exist on the file system of the indexing node.  As with the INCOMING FILE command, the indexing node  streams each index file after sending an INDEX FILE command.  The INDEX FILE command has two parameters: the first being  the name of the index, and the second is the size of the file in  bytes.  SEND COMPLETE: When sending the sets of files for both the  index and the files to be indexed, the node must notify the  corresponding node when the process is complete. Once the  initiating node is finished sending the set of documents to be  indexed, it will then send a SEND COMPLETE command  indicating to the indexing node that there are no more files and  the node can proceed with indexing the files. In the case of the  initiating node sending the index files, the indexing node will  complete the transfer with the SEND COMPLETE command  indicating to the initiating node that there are no more index files  to be sent and the initiating node can then assemble those index  files into the main index.  The NodeController is responsible for setting up connections with  nodes in the idle state to distribute the indexing process. Using  JXTA [5], the node controller will obtain a set of nodes. This set  of nodes is iterated and each one is sent the IDLE QUERY  command. The nodes that respond with idle are then collected.  The set of idle nodes includes the node initiating the distributed  indexing process, referred to as the local node. Once the  collection of idle nodes is obtained, the node updates the set of  controllers and evenly divides the set of documents that are to be  indexed. For example, if there are 100 documents and 10 nodes  (including the local node) then each node will have 10 documents  to index. For each indexing node an instance of the FileSender  object is created. The FileSender is aware of the set of documents  that node is responsible for. Once a FileSender object has been  created for each node, the NodeController waits for each  FileSender to complete. When the FileSender objects have  completed the NodeController will take the resultant indexes from  175  each node and pass them to an instance of the IndexCompiler,  which maintains the index and the list of FileSenders. Once the  IndexCompiler has completed it will return to the idle state and  activate the directory scanner to monitor the locally owned set of  documents for changes that may require reindexing.  The NodeIndexer is responsible for receiving documents sent to it  by the initiating node and then indexing them using the Lucene  engine [7]. Once the indexing is complete the resulting index is  streamed back to the initiating node as well as compiled in the  indexer nodes own local index. Before initiating the indexing  process it must be sent an IDLE QUERY message. This is the  first command that sets off the indexing process. The indexer  node will determine whether it is considered idle based on the  current CPU usage. As outlined in the protocol section if the node  is not being used and has a low overall CPU usage percentage it  will return IDLE to the IDLE QUERY command. If the indexer  nodes CPU usage is above 50% for a specified amount of time it  is then considered to be busy and will respond to the IDLE  QUERY command with BUSY. If a node is determined busy it  returns to its listening state waiting for another IDLE QUERY  from another initiating node. If the node is determined to be idle it  will enter the state where it will receive files from the initiating  node that it is responsible for indexing. Once all of the files are  received by the initiating node, indicated by a SEND  COMPLETE message, it starts an instance of the Lucene indexing  engine. The files are stored in a temporary directory separate from  the nodes local documents that it is responsible for maintaining an  index of. The Lucene index writer then indexes all of the  transferred files. The index is stored on the drive within a  temporary directory separate from the current index. After the  indexing of the files completes the indexer node enters the state  where the index files are sent back to the initiating node. The  indexer node loops through all of the files created by Lucene"s  IndexWriter and streams them to the initiating node. Once these  files are sent back that index is then merged into the indexer  nodes own full index of the existing files. It then enters the idle  state where it will then listen for any other nodes that required  distributing the indexing process.  The FileSender object is the initiating node equivalent of the  indexer node. It initiates the communication between the initiating  node and the node that will assist in the distributed indexing. The  initiating node runs many instances of the FileSender node one  for each other node it has determined to be idle. Upon  instantiation of the FileSender it is passed the node that it is  responsible for contacting and the set of files that must be sent.  The FileSender"s first job is to send the files that are to be indexed  by the other idle node. The files are streamed one at a time to the  other node. It sends each file using the INCOMING FILE  command. With that command it sends the name of the file being  sent and the size in bytes. Once all files have been sent the  FileSender sends the SEND COMPLETE command. The  FileSender creates an instance of Lucene"s IndexWriter and  prepares to create the index in a temporary directory on the file  system. The FileSender will begin to receive the files that are to  be saved within the index. It receives an INDEX FILE command  with the name of the files and the size in bytes. This file is then  streamed into the temporary index directory on the FileSender  node. After the transfer of the index files has been completed the  FileSender notifies the instance of the index compiler that it is  ready to combine the index. Each instance of the FileSender has  its own unique section of temporary space to store the index that  has been transferred back from the indexing node. When  notifying the IndexCompiler it will also pass the location of the  particular FileSenders directory location of that index.  4. PEER-TO-PEER DISTRIBUTION  Apocrita uses a peer-to-peer distribution model in order to  distribute files. Files are distributed solely from a serving node to  a client node without regard for the availability of file pieces from  other clients in the network. This means that the file transfers will  be fast and efficient and should not severely affect the usability of  serving nodes from the point of view of a local user. The JXTA  framework [5] is used in order to implement peer-to-peer  functionality. This has been decided due to the extremely   shorttimeline of the project which allows us to take advantage of over  five years of testing and development and support from many  large organizations employing JXTA in their own products. We  are not concerned with any potential quality problems because  JXTA is considered to be the most mature and stable peer-to-peer  framework available.  Using JXTA terminology, there are three types of peers used in  node classification.  Edge peers are typically low-bandwidth, non-dedicated nodes.  Due to these characteristics, edge peers are not used with  Apocrita.  Relay peers are typically higher-bandwidth, dedicated nodes.  This is the classification of all nodes in the Apocrita network, and,  as such, are the default classification used.  Rendezvous peers are used to coordinate message passing  between nodes in the Apocrita network. This means that a  minimum of one rendezvous peer per subnet is required.  4.1 Peer Discovery  The Apocrita server subsystem uses the JXTA Peer Discovery  Protocol (PDP) in order to find participating peers within the  network as shown in Figure 2.  Figure 2. Apocrita peer discovery process.  176  The PDP listens for peer advertisements from other nodes in the  Apocrita swarm. If a peer advertisement is detected, the server  will attempt to join the peer group and start actively contributing  to the network. If no peers are found by the discovery service, the  server will create a new peer group and start advertising this peer  group. This new peer group will be periodically advertised on the  network; any new peers joining the network will attach to this  peer group. A distinct advantage of using the JXTA PDP is that  Apocrita does not have to be sensitive to particular networking  nuances such as Maximum Transmission Unit (MTU). In  addition, Apocrita does not have to support one-to-many packet  delivery methods such as multicast and instead can rely on JXTA  for this support.  4.2 Index Query Operation  All nodes in the Apocrita swarm have a complete and up-to-date  copy of the network index stored locally. This makes querying the  index for search results trivial. Unlike the Gnutella protocol, a  query does not have to propagate throughout the network. This  also means that the time to return query results is very fast - much  faster than protocols that rely on nodes in the network to pass the  query throughout the network and then wait for results. This is  demonstrated in Figure 3.  Figure 3. Apocrita query operation.  Each document in the swarm has a unique document  identification number (ID). A node will query the index and a  result will be returned with both the document ID number as well  as a list of peers with a copy of the matched document ID. It is  then the responsibility of the searching peer to contact the peers in  the list to negotiate file transfer between the client and server.  5. PROTOTYPE IMPLEMENTATION  Apocrita uses the Lucene framework [7], which is a project under  development by the Apache Software Foundation. Apache  Lucene is a high-performance, full-featured text search engine  library written entirely in Java. In the current implementation,  Apocrita is only capable of indexing plain text documents.  Apocrita uses the JXTA framework [5] as a peer-to-peer transport  library between nodes. JXTA is used to pass both messages and  files between nodes in the search network. By using JXTA,  Apocrita takes advantage of a reliable, and proven peer-to-peer  transport mechanism. It uses the pipe facility in order to pass  messages and files between nodes. The pipe facility provides  many different types of pipe advertisements. This includes an  unsecured unicast pipe, a secured unicast pipe, and a propagated  unsecured pipe.  Message passing is used to pass status messages between nodes in  order to aid in indexing, searching, and retrieval. For example, a  node attempting to find an idle node to participate in indexing will  query nodes via the message facility. Idle nodes will reply with a  status message to indicate they are available to start indexing.  File passing is used within Apocrita for file transfer. After a file  has been searched for and located within the peer group, a JXTA  socket will be opened and file transfer will take place. A JXTA  socket is similar to a standard Java socket, however a JXTA  socket uses JXTA pipes in underlying network transport. File  passing uses an unsecured unicast pipe in order to transfer data.  File passing is also used within Apocrita for index transfer. Index  transfer works exactly like a file transfer. In fact, the index  transfer actually passes the index as a file. However, there is one  key difference between file transfer and index transfer. In the case  of file transfer, a socket is created between only two nodes. In the  case of index transfer, a socket must be created between all nodes  in the network in order to pass the index, which allows for all  nodes to have a full and complete index of the entire network. In  order to facilitate this transfer efficiently, index transfer will use  an unsecured propagated pipe to communicate with all nodes in  the Apocrita network.  6. PERFORMANCE EVALUATION  It is difficult to objectively benchmark the results obtained  through Apocrita because there is no other system currently  available with the same goals as Apocrita. We have, however,  evaluated the performance of the critical sections of the system.  The critical sections were determined to be the processes that are  the most time intensive. The evaluation was completed on  standard lab computers on a 100Mb/s Ethernet LAN; the  machines run Windows XP with a Pentium 4 CPU running at  2.4GHz with 512 MB of RAM.  The indexing time has been run against both: the Time Magazine  collection [8], which contains 432 documents and 83 queries and  their most relevant results, and the NPL collection [8] that has a  total of 11,429 documents and 93 queries with expected results.  Each document ranges in size between 4KB and 8KB. As Figure  4 demonstrates, the number of nodes involved in the indexing  process affects the time taken to complete the indexing   processsometimes even drastically.  Figure 4. Node vs. index time.  The difference in going from one indexing node to two indexing  nodes is the most drastic and equates to an indexing time 37%  faster than a single indexing node. The different between two  177  indexing nodes and three indexing nodes is still significant and  represents a 16% faster time than two indexing nodes. As the  number of indexing nodes increases the results are less dramatic.  This can be attributed to the time overhead associated with having  many nodes perform indexing. The time needed to communicate  with a node is constant, so as the number of nodes increases, this  constant becomes more prevalent. Also, the complexity of joining  the indexing results is a complex operation and is complicated  further as the number of indexing nodes increases.  Socket performance is also a very important part of Apocrita.  Benchmarks were performed using a 65MB file on a system with  both the client and server running locally. This was done to  isolate possible network issues. Although less drastic, similar  results were shown when the client and server run on independent  hardware. In order to mitigate possible unexpected errors, each  test was run 10 times.  Figure 5. Java sockets vs. JXTA sockets.  As Figure 5 demonstrates, the performance of JXTA sockets is  abysmal as compared to the performance of standard Java sockets.  The minimum transfer rate obtained using Java sockets is  81,945KB/s while the minimum transfer rater obtained using  JXTA sockets is much lower at 3, 805KB/s. The maximum  transfer rater obtain using Java sockets is 97,412KB/s while the  maximum transfer rate obtained using JXTA sockets is  5,530KB/s. Finally, the average transfer rate using Java sockets is  87,540KB/s while the average transfer rate using JXTA sockets is  4,293KB/s.  The major problem found in these benchmarks is that the  underlying network transport mechanism does not perform as  quickly or efficiently as expected. In order to garner a  performance increase, the JXTA framework needs to be  substituted with a more traditional approach. The indexing time is  also a bottleneck and will need to be improved for the overall  quality of Apocrita to be improved.  7. RELATED WORK  Several decentralized P2P systems [1, 2, 3] exist today that  Apocrita features some of their functionality. However, Apocrita  also has unique novel searching and indexing features that make  this system unique. For example, Majestic-12 [4] is a distributed  search and indexing project designed for searching the Internet.  Each user would install a client, which is responsible for indexing  a portion of the web. A central area for querying the index is  available on the Majestic-12 web page. The index itself is not  distributed, only the act of indexing is distributed. The distributed  indexing aspect of this project most closely relates Apocrita goals.  YaCy [6] is a peer-to-peer web search application. YaCy consists  of a web crawler, an indexer, a built-in database engine, and a p2p  index exchange protocol. YaCy is designed to maintain a  distributed index of the Internet. It used a distributed hash table  (DHT) to maintain the index. The local node is used to query but  all results that are returned are accessible on the Internet. YaCy  used many peers and DHT to maintain a distributed index.  Apocrita will also use a distributed index in future  implementations and may benefit from using an implementation  of a DHT. YaCy however, is designed as a web search engine  and, as such solves a much different problem than Apocrita.  8. CONCLUSIONS AND FUTURE WORK  We presented Apocrita, a distributed P2P searching and indexing  system intended for network users on an Intranet. It can help  organizations with no network file server or necessary network  infrastructure to share documents. It eliminates the need for  documents to be manually shared among users while being edited  and reduce the possibility of conflicting versions being  distributed. A proof of concept prototype has been constructed,  but the results from measuring the network transport mechanism  and the indexing time were not as impressive as initially  envisioned. Despite these shortcomings, the experience gained  from the design and implementation of Apocrita has given us  more insight into building challenging distributed systems.  For future work, Apocrita will have a smart content distribution  model in which a single instance of a file can intelligently and  transparently replicate throughout the network to ensure a copy of  every important file will always be available regardless of the  availability of specific nodes in the network. In addition, we plan  to integrate a revision control system into the content distribution  portion of Apocrita so that users could have the ability to update  an existing file that they found and have the old revision  maintained and the new revision propagated. Finally, the current  implementation has some overhead and redundancy due to the  fact that the entire index is maintained on each individual node,  we plan to design a distributed index.  9. REFERENCES  [1] Rodrigues, R., Liskov, B., Shrira, L.: The Design of a Robust  Peer-to-Peer System. Available online:  http://www.pmg.lcs.mit.edu/~rodrigo/ew02-robust.pdf.  [2] Chawathe, Y., Ratnasamy, S., Breslau, L., Lanham, N., and  Chenker, S.: Making Gnutella-like P2P Systems Scalable. In  Proceedings of SIGCOMM"03, Karlsruhe, Germany.  [3] Harvest: A Distributed Search System:  http://harvest.sourceforge.net.  [4] Majestic-12: Distributed Search Engine:  http://www.majestic12.co.uk.  [5] JXTA: http://www.jxta.org.  [6] YaCy: Distributed P2P-based Web Indexing:  http://www.yacy.net/yacy.  [7] Lucene Search Engine Library: http://lucene.apache.org.  [8] Test Collections (Time Magazine and NPL):  www.dcs.gla.ac.uk/idom/ir_resources/test_collections.  178