require 'parallel'

parallel.fork()
parallel.nfork(4)

parallel.nfork( {4, ip='myserver.org', protocol='ssh', lua='/path/to/remote/torch'},
                {6, ip='myserver2.org', protocol='ssh', lua='/path/to/remote/torch'} )

parallel.addremote( {ip='server1.org', cores=8, lua='/path/to/torch', protocol='ssh -Y'},
                    {ip='server2.org', cores=16, lua='/path/to/torch', protocol='ssh -Y'},
                    {ip='server3.org', cores=4, lua='/path/to/torch', protocol='ssh -Y'} )
parallel.sfork(16)

-- in this example, the 16 processes will be distributed over the 3 machines:
-- server1.org: 6 processes
-- server2.org: 6 processes
-- server3.org: 4 processes

parallel.addremote(...)
parallel.calibrate()
forked = parallel.sfork(parallel.remotes.cores)  -- fork as many processes as cores available
for _,forked in ipairs(forked) do
   print('id: ' .. forked.id .. ', speed = ' .. forked.speed)
end
-- the speed of each process is a number ]0..1]. A coef of 1 means that it is the
-- fastest process available, and 0.5 for example would mean that the process is 2x
-- slower

-- define process' code:
code = function()
   -- arbitrary code contained here
   require 'torch'
   t = torch.Tensor(10)
   print(t)

   -- any process can access its id, its parent's id [and children's id]
   print(parallel.id)
   print(parallel.parent.id)
   if parallel.children[1] then print(parallel.children[1].id) end

   -- if arguments were passed, they're found in the regular ... table        
   args = {...}    
   print(args[1])
end

-- execute code in given process(es), with optional arguments:
parallel.children:exec(code)

-- this is equivalent to:
for _,child in ipairs(parallel.child) do
    child:exec(code)
end

-- child code:
code = function()
   while true do
      print('something')
      parallel.yield()
   end
end
c = parallel.fork()
c:exec(code)

-- parent code
for i = 1,10 do
    c:join()
end

-- each time join() is called, it waits for the child to yield, and vice-versa.
-- in that example, 'something' only gets printed when the parent joins its child

-- child code:
code = function()
   while true do
      print('something')
      m = parallel.yield()
      if m == 'break' then break end
   end
end
c = parallel.fork()
c:exec(code)

-- parent code
c:join('break')

code = function()
     -- do nothing and die
end
parallel.nfork(1)              -- fork one process
parallel.children:exec(code)   -- execute dummy code
print(parallel.nchildren)      -- prints: 1
parallel.children:sync()       -- wait for all children (here only 1) to die
print(parallel.nchildren)      -- prints: 0
parallel.nfork(2)              -- fork 2 processes
print(parallel.nchildren)      -- prints: 2
print(parallel.children[1])    -- prints: nil
print(parallel.children[2])    -- prints: table --- current running processes always
print(parallel.children[3])    -- prints: table --- exist in children[process.id]

-- define some code for children
somecode = function()
   while true do
      -- in an infinite loop, receive objects from parent:
      local obj = parallel.parent:receive()
      -- print
      parallel.print('received object:', obj)
   end
end

-- dispatch two processes:
parallel.nfork(2)
parallel.children:exec(somecode)

-- and send them some data:
t = {'a table', entry2='with arbitrary entries', tensor=torch.Tensor(100,100)}
while true do
    parallel.children[1]:send(t)        -- send the whole table to child 1
    parallel.children[2]:send(t.entry2) -- just send an entry to child 2
end

> parallel.print('something')

<parallel#014>  something

worker = function()
       -- some worker code
end

parent = function()
       -- some parent code
end

ok,err = pcall(parent)
if not ok then
   print(err)
   parallel.close()   -- this is the key call: doing this will insure leaving a clean
                      -- state, whatever the error was (ctrl+c, internal error, ...)
end

-- required libs
require 'parallel'

-- define code for workers:
function worker()
   -- a worker starts with a blank stack, we need to reload
   -- our libraries
   require 'sys'
   require 'torch'

   -- print from worker:
   parallel.print('Im a worker, my ID is: ' .. parallel.id .. ' and my IP: ' .. parallel.ip)

   -- define a storage to receive data from top process
   while true do
      -- yield = allow parent to terminate me
      m = parallel.yield()
      if m == 'break' then break end

      -- receive data
      local t = parallel.parent:receive()
      parallel.print('received object with norm: ', t.data:norm())

      -- send some data back
      parallel.parent:send('this is my response')
   end
end

-- define code for parent:
function parent()
   -- print from top process
   parallel.print('Im the parent, my ID is: ' .. parallel.id)

   -- fork N processes
   parallel.nfork(4)

   -- exec worker code in each process
   parallel.children:exec(worker)

   -- create a complex object to send to workers
   t = {name='my variable', data=torch.randn(100,100)}

   -- transmit object to each worker
   parallel.print('transmitting object with norm: ', t.data:norm())
   for i = 1,1000 do
      parallel.children:join()
      parallel.children:send(t)
      replies = parallel.children:receive()
   end
   parallel.print('transmitted data to all children')

   -- sync/terminate when all workers are done
   parallel.children:join('break')
   parallel.print('all processes terminated')
end

-- protected execution:
ok,err = pcall(parent)
if not ok then print(err) parallel.close() end