13-Parallel Job and File Distribution in an Agent Hierarchy

1 .Parallel Job and File Distribution in an Agent Hierarchy Munehiro Fukuda Computing & Software Systems, University of Washington, Bothell Funded by CSS434: Parallel and Distributed 1 Computing 

2 .Outline 1. Objectives 2. AgentTeamwork 3. Job Distribution 4. File Distribution 5. Performance Evaluation 6. Summary CSS434: Parallel and Distributed 2 Computing 

3 .Objectives  Utilizing Campus Computing Resources  Why Mobile Agents?  Objectives CSS434: Parallel and Distributed 3 Computing 

4 . 1. Objective Utilizing Campus Computing s Resources Labs at 3pm on a weekday  Computational demands:  Research: UWB Brain Network, etc.  Teaching: Parallel & Distributed Comp, Network Design, etc.  Resources set-up and management:  Instructional machines renewed periodically and maintained  Research machines purchased with external funds asynchronously  Needs for allocating idle machines to demanding applications CSS434: Parallel and Distributed 4 Computing 

5 . 1. Objective Why Mobile Agents? s  An execution model previously highlighted as a prospective infrastructure of distributed systems.  Static job deployment and result collection: No more than an alternative approach to centralized grid middleware implementation  Our goal: Use agents’ navigational/behavioral autonomy for grid computing Central FTP Cycle Cycle Cycle Server HTTP Server manger RPC Server User Internet CSS434: Parallel and Distributed 5 Computing 

6 . 1. Objective Objectives s  Multi-cluster job coordination  Use of mobile agents  Supporting clusters of  Migrating and resuming jobs asynchronously-controlled, with agents independent computing nodes  Check-pointing various  Having each agent check- pointing its corresponding communication styles in user process applications: master slave, heart beat, all reduce..  Improving performance for job  Performing in a hierarchy of deployment, file transfer, mobile agents monitoring, and resumption CSS434: Parallel and Distributed 6 Computing 

7 .AgentTeamwork  System Overview  Execution Layer  Top-Down Details of Each Layer  User Program Wrapper  Error Recoverable Multi-Cluster Tcp  UWAgent Mobile Agent Platform CSS434: Parallel and Distributed 7 Computing 

8 . 2. AgentTeamwor System Overview k User A’s User A’s User B’s Process Process Process TCP Communication Snapshot Snapshot Snapshot GridTCP GridTCP GridTCP Methods Methods Methods Results User program wrapper User program wrapper User program wrapper Results snapshot snapshot snapshot Sentinel Sentinel Sentinel Agent Agent Agent Commander Resource Resource Commander Agent Agent Agent Agent User User User A B B FTP Server snapshots snapshots Bookkeeper CSS434: Parallel and Distributed Bookkeeper 8 Agent Agent Computing 

9 . 2. AgentTeamwor Execution Layer k Java user applications mpiJava API mpiJava-S mpiJava-A Java socket GridTcp User program wrapper Commander, resource, sentinel, and bookkeeper agents UWAgents mobile agent execution platform Operating systems CSS434: Parallel and Distributed 9 Computing 

10 . 2. AgentTeamwor User Program and Wrapper k 1 import AgentTeamwork.Ateam.*; 2 public class MyApplication extends AteamProg { 1 import AgentTeamwork.Ateam.*; 1 import AgentTeamwork.Ateam.*; 3 public MyApplication( Object o ){} 2 public class MyApplication extends AteamProg { 2 public class MyApplication extends AteamProg { 4 public MyApplication( ) {} 3 public MyApplication( Object o ){} 3-4 .....; 5 private void compute( ) { 4 public MyApplication( ) {} 5 private void compute( ) { 6 if ( MPI.rank( ) == 0 ) backup 5 private void compute( ) { 6 if ( MPI.rank( ) == 0 ) 7 MPI.Send( ); 6 if ( MPI.rank( ) == 0 ) 7 MPI.Send( ); 8 else ifbackup ( MPI.rank( ) == 1 ) 7 MPI.Send( ); 8 else if ( MPI.rank( ) == 1 ) 9 MPI.Recv( ); incoming backup incoming 8 else if ( MPI.rank( ) == 1 ) 9 MPI.Recv( ); 10 ateam.takeSnapshot( phase ); 9 MPI.Recv( ); incoming 10 ateam.takeSnapshot( phase ); 11 } 10 ateam.takeSnapshot( phase ); 11 } 12 public static void main( String[] args ) { 11 } 12 public static void main( String[] args ) { 13 MyApplication program = null; 12 public static void main( String[] args ) { 13-21 .....; 14 if ( ateam.isResumed( ) } 13 MyApplication program = null; 22 prgoram.compute( ); 15 program = ( MyApplication ) 14 if ( ateam.isResumed( ) } 23 MPI.Finalize( args ); 16 ateam.retrieveLocalvar( “program” ); 15 program = ( MyApplication ) 24 } 17 } else { 16 ateam.retrieveLocalvar( “program” ); 25 } User Program Wrapper 18 MPI.Init( args ); 17 } else { 19 program = new MyApplication( ); 18 MPI.Init( args ); 20 ateam.registerLocalVar(“program”, program); 19 program = new MyApplication( ); 21 } 20 ateam.registerLocalVar(“program”, program); 22 prgoram.compute( ); 21 } 22 prgoram.compute( ); 23 24 } MPI.Finalize( args ); n1.uwb.edu 23 MPI.Finalize( args ); User Program Wrapper 25 } 24 } Snapshot User Program Wrapper 25 } maintenance n0.uwb.edu n2.uwb.edu CSS434: Parallel and Distributed 10 Computing 

11 . 2. AgentTeamwor Error-Recoverable k MultiCluster Tcp User Program Wrapper User Program Wrapper User Program Wrapper rank dest gateway rank dest gateway rank dest gateway 0 perseus - 0 perseus - 0 perseus - 1 mnode1 mnode0 - 1 uw1-320-01 - 1 mnode1 uw1-320-01 mnode0 metis medusa 2 medusa - 2 metis - 2 medusa umetis - user user user User Program Wrapper program program program User Program Wrapper rank dest gateway rank 2 rank 2 rank 0 0 rank perseus dest medusa gateway Internet 0 perseus medusa meits 1 mnode0 - 1 uw1-320-01 mnode1 - 2 medusa - 2 medusa metis - medusa.uwb.edu user program user program rank 1 metis.uwb.edu rank 1 perseus.uwb.edu cluster-private domain uw1-320-00 mnode1 mnode0 uw1-320-01 uw1-320-04 mnode4 CSS434: Parallel and Distributed 11 Computing 

12 . 2. AgentTeamwor UWAgent Mobile Agent k Platform UWInject: submits a new agent from shell. User Agent hierarchy (time=3:30pm, 8/25/05 ip = medusa.uwb.edu id 0 id 0 -m 4 -m 3 name = fukuda) id 1 id 1 id 2 id 3 id 2 A user job UWPlace Agent hierarchy (time=3:31pm, 8/25/05 id 12 ip = perseus.uwb.edu id 4 id 5 id 6 id 7 id 8 id 9 id 10 id 11 name = fukuda)  An agent hierarchy created per each submission from the Unix shell  Messages forwarded through an agent tree (or directly if possible)  A user job scheduled as a thread, using suspend/resume CSS434: Parallel and Distributed 12 Computing 

13 .Job Distribution  Job Deployment in Public Domain  Challenges in Multi-Cluster Agent Tree  Multi-Cluster Job Deployment  Multi-Cluster Job Resumption CSS434: Parallel and Distributed 13 Computing 

14 . 3. Job Distribution Job Deployment in Public Job Submission Domain User Commander id 0 XML Query Spawn Sentinel XML Resource Bookkeeper id 2 DB id 1 id 3 rank 0 Sentinel Sentinel Sentinel Sentinel Sensor Sensor Bookkeeper Bookkeeper id 8 id 9 id 10 id 11 id 4 id 5 id 12 id 13 rank 1 rank 2 rank 3 rank 4 snapshot Sentinel Sentinel Sentinel id: agent id id 32 id 33 id 34 rank: MPI Rank rank 5 rank 6 rank 7 snapshot CSS434: Parallel and Distributed 14 Computing 

15 . 3. Job Distribution Challenges in Multi-Cluster Agent Tree Commander User Partition 1 2 id 0 Sentinel Resource Bookkeeper id 2 id 1 id 3 rank 0 Sentinel Cluster 0 Sentinel Sentinel Sentinel id 8 id 9 id 10 id 11 rank 1 rank 2 rank 3 rank 4 Cluster 2 Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 32 id 33 id 34 id 35 Cluster 1 id 46 id 47 rank 5 rank 6 rank 7 rank 8 rank 19 rank 20 CSS434: Parallel and Distributed 15 Computing 

16 . 3. Job Distribution Multi-Cluster Job Deployment Commander User id 0 Sentinel id 2 Resource rank -1 Bookkeeper id 1 Cluster gateway 0 id 3 Desktop computers Sentinel Sentinel id 8 id 9 rank -2 Cluster gateways 1, 2, and 3 rank X Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 32 id 33 id 34 id 35 id 36 id 37 id 38 id 39 rank 0 rank -3 rank -4 rank -5 rank X+1 rank X+2 rank X+3 rank X+4 Sentinel Sentinel Sentinel Sentinel Sentinel id 128 id 129 id 130 id 131 id 132 Cluster 1 rank 1 rank 2 rank 3 rank 4 rank 6 Cluster 3 Cluster 0 Cluster 2 Sentinel Sentinel Sentinel Sentinel Sentinel id 512 id 528 id 529 id 530 id 531 rank 5 rank 7 rank 8 rank 9rank 10 CSS434: Parallel and Distributed 16 Computing 

17 . 3. Job Distribution Multi-Cluster Job Resumption Commander User id 0 Sentinel id 2 Resource Cluster gateway 0 rank -1 Desktop computers Bookkeeper id 1 id 3 Sentinel Sentinel id 8 id 9 rank -2 rank 11 Cluster gateways 1, 2, and 3 Sentinel Sentinel Sentinel Sentinel Sentinel id 32 id 32 id 33 id 34 id 35 rank 0 Cluster 0 rank 0 Another Cluster 0 rank -3 rank -4 rank -5 Sentinel Sentinel Sentinel SentinelSentinel Sentinel Sentinel Sentinel Sentinel id 128 id 129 id 130 id 131 id 128 id 129 id 130 id 131 id 132 Cluster 1 rank 1 rank 2 rank 3 rank 4 rank 1 rank 2 rank 3 rank 4 rank 6 Cluster 3 Cluster 2 Sentinel New Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 512 id 512 id 512 id 528 id 529 id 530 id 531 rank 5 rank 5 rank 5 rank 7 rank 8 rank 9rank 10 CSS434: Parallel and Distributed 17 Computing 

18 .File Distribution  File Duplication in a Hierarchy  File Partitioning  File Fragmentation and Aggregation  File-Strip Check-Pointing and Recovery CSS434: Parallel and Distributed 18 Computing 

19 .File Duplication in a Hierarchy 1. A commander agent reads a file commander 2. A sentinel agent duplicates the file file (1) read a file 3. The sentinel agent sends the file to its child agents sentinel file (2) duplicate the file sentinel sentinel sentinel sentinel file (3) transfer the files sentinel sentinel sentinel sentinel CSS434: Parallel and Distributed 19 Computing 

20 . 4. File Distribution File Partitioning MPI/IO-based file  etype  Divide a unitanofentire file access fileainto multiple primitive data type stripes.  filetype Specify which file portion a is repetitive file record delivered a given MPI title with etypes rank.  Define an etype and a filetype of each file portion. CSS434: Parallel and Distributed 20 Computing 

21 . 4. File Distribution File Fragmentation and Aggregation key value read files 128_inputFile1_1 contents GU commander 52 Id: 0 8 52 32_inputFile1_0 contents I 8 32_inputFile2_0 contents 128_inputFile1_1 contents 528_inputFile2_7 contents root 528_inputFile1_7 contents 32_inputFile1_0 contents sentinel 32_inputFile2_0 contents Id: 2 32 528_inputFile2_7 contents 128 sentinel 528 32 128 528 sentinel 528_inputFile1_7 contents Id: 8 Id: 9 sentinel sentinel sentinel sentinel sentinel sentinel Id: 32 Id: 33 Id: 36 Id: 37 Id: 38 Id: 39 sentinel sentinel sentinel sentinel sentinel Id: 128 Id: 129 Id: 130 Id: 131 Id: 132 128_inputFile1_1 contents 32_inputFile1_0 contents sentinel CSS434: 32_inputFile2_0 Id:528Parallel and Distributed contents 21 Computing 

22 . File-Stripe Check-Pointing and Recovery Messages relayed through  commander a tree Message resend  Thread threads created within each sentinel bookkeeper  Strips passed through the sentinel retrieval main (user) thread  Outputs directly sent back Input thread check-pointing to the commander sentinel  Snapshot taken and reported to a sentinel main thread  Snapshot retrieved upon a crash sentinel output thread  Loss messages requested directly from the commander and the source agent CSS434: Parallel and Distributed 22 Computing 

23 .Performance  Multi-Cluster Job Deployment  Multi-Cluster Job Check-Pointing  Comparison of File Duplication  File Fragmentation and Pipelined Transfer  Random-Access File Transfer  File Recovery Overheads CSS434: Parallel and Distributed 23 Computing 

24 . 5. Performanc Multi-Cluster Job e Deployment Depth First cmd AgentTeamwork duroc Duroc/OpenPBS 100Mbps backbone Snt Snt PBS PBS 8 32 ….. ….. Cluster-R Cluster-I (32CPUs 2.8-3.2GHz 1Gpbs) (32CPUs 1.5GHz 100M-1Gpbs) cmd duroc Breath First Snt Snt PBS PBS 8 32 CSS434: Parallel and Distributed 24 Computing 

25 . 5. Performanc Multi-Cluster Job Check- e Pointing Commander User id 0 Sentinel id 2 Bookkeeper Resource rank -1 id 3 id 1 Sentinel Cluster R Gateway id 8 Bookkeeper rank -2 id 12 Sentinel Sentinel id 32 id 33 Cluster I Gateway rank 0 Cluster R rank -3 Sentinel Sentinel Sentinel Sentinel Sentinel id 128 id 129 id 130 id 131 id 132 rank 1 rank 2 rank 3 rank 4 rank 32 Cluster I Sentinel id 512 ….. Sentinel Sentinel Sentinel Sentinel id 528 id 529 id 530 id 531 rank 5 rank 33rank 34rank 35rank 36 CSS434: Parallel and Distributed 25 Computing 

26 . 5. Performanc Comparison of File e Duplication Commander User id 0 sentinel id 2 rank 0 sentinel id 9 rank 0 Sentinel Sentinel Sentinel Sentinel id 36 id 37 id 38 id 39 rank 1 rank 2 rank 3 rank 4 Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 144 id 145 id 146 id 147 id 156 id 157 rank 5 rank 6 rank 7 rank 8 rank 19 rank 20 CSS434: Parallel and Distributed 26 Computing 

27 . 5. Performanc File Fragmentation and e Pipeline Transfer Commander User 1M 1M read read 1M read id 0 1M read 1M read sentinel id 2 rank 0 sentinel 1M 1M id1M9 1M rank 1M 0 1M Sentinel 1M Sentinel Sentinel Sentinel 1M 1M 1M id 1M36 id 37 id 38 id 39 1M 1M 1M rank 1M 1 rank 2 rank 3 rank 4 Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 144 id 145 id 146 id 147 id 156 id 157 rank 5 rank 6 rank 7 rank 8 rank 19 rank 20 CSS434: Parallel and Distributed 27 Computing 

28 . 5. Performanc Random-Access File Transfer e CSS434: Parallel and Distributed 28 Computing 

29 . 5. Performanc File Recovery Overheads e User Test cases File Commander distribution 1M time 1M read read 1M read id 0 1M read Test1: no agent termination 262.60 sec 1M read Test2: rank-0 termination 293.57 sec sentinel Test3: rank-11 termination id 2 266.68 sec rank 0  Recovery overhead atida 9 leaf: 4.08 sec sentinel 1M 1M 1M 1M rank 1M 0  Recovery overheads at the root:  30.97sec for 11 nodes (2.82 per node) 1M Sentinel1M Sentinel Sentinel Sentinel 1M  87.28sec id 36 for 31 nodes 1M 1M id 37 id 38 id 39 1M 1M 1M 1M rank 1 1M rank 2 rank 3 rank 4  177.4sec for 63 nodes Sentinel Sentinel Sentinel Sentinel Sentinel Sentinel id 144 id 145 id 146 id 147 id 156 id 157 rank 5 rank 6 rank 7 rank 8 rank 19 rank 20 CSS434: Parallel and Distributed 29 Computing