Thank you for your reply.
Your understanding of my requirement sounds about right. However I had
tested that same approach as you suggested (passing arguments strings and
lists to the server) and having the server instantiate the class and
execute the process. (see code below with both exposed functions) However I
was getting the same error on both approaches.
If I understand your question regarding "creating Job object on client and
not using it" correctly - the reason I was creating the Job() on the client
is only because the details for the Job and output & return code from the
execution of the Job() is in the Job() object for handling on the client
side, and I do use it in the client side by passing it to the exposed
service on the server. I could approach this differently as you mention by
passing in the respective strings and create on the server and retrieve
return code and output in a different way, but that doesn't resolve my
error.
This is the code with both approaches
*SERVER CODE:*
import os
import job
import multiprocessing
import rpyc
from rpyc.utils.server import ThreadedServer
######
class MyClass(rpyc.Service):
""" Manages jobs requested on a Node instance"""
def on_connect(self):
# code that runs when a connection is created
# (to init the serivce, if needed)
pass
def on_disconnect(self):
# code that runs when the connection has already closed
# (to finalize the service, if needed)
pass
def exposed_submitJobObj(self,newJob):
"""Create a new Job using the input arguments"""
self.addJobToQueue(newJob)
def exposed_submitJob(self,env,jobType,jobcmd,arglst):
"""Create a new Job using the input arguments"""
newjob = job.Job(jobcmd,arglst)
jobwrapper = job.JobWrapper()
jobwrapper.setJobAttributes(env,newjob)
self.addJobToQueue(jobwrapper)
def addJobToQueue(self,newJob):
"""Adds a new job to the node job execution queue and executes it"""
#execute the job in a new thread
myJobWorker = multiprocessing.Process(target=self.jobWorker, args=(
newJob,))
myJobWorker.start()
def jobWorker(self,currJob):
"""Multiprocess method to run job as a parallel process """
currJob.execute()
def startNode():
#retrieve node name from node configuration file
appConfig = util.config()
t = ThreadedServer(MyClass, port = 18861, protocol_config = {
"allow_public_attrs" : True,"allow_pickle":True})
t.start()
######
if __name__ == '__main__':
startNode()
*CLIENT CODE:*
import os
import job
import rpyc
def testJobWrapper():
c = rpyc.connect("localhost", 18861, config = {"allow_public_attrs" :
True,"allow_pickle":True})
c.root.exposed_test()
winenv='C:\\py_sandbox\\schedulex_scripts\\test.env'
wintype='Job'
winarg=['param1']
wincmd = 'C:\\py_sandbox\\schedulex_scripts\\hellow.bat'
# ---->APPROACH 1 - using job()) <----
winJob = job.Job(wincmd,winarg)
testJob = job.JobWrapper()
testJob.setJobAttributes('C:\\py_sandbox\\schedulex_scripts\\test.env',
winJob)
c.root.exposed_submitJobObj(testJob)
#---->APPROACH 2<------
c.root.exposed_submitJob(winenv,wintype,wincmd,winarg)
if __name__ == '__main__':
testJobWrapper()
FYI: This is Python 2.7 running on Windows,
Thanks
On Monday, September 21, 2015 at 9:47:48 PM UTC+8, Rüdiger Kessel wrote:
>
> Hi,
>
> your example suggests that the program hellow.bat is execute on the server.
>
> The filepath of hellow.bat is probably a local path on the server.
>
> You create Job object on the client side and you pass this Job object to
> the server.
>
> How does the Job object (on the client side) know that the job is intended
> to be run on the server instead on the client where the job is created?
>
> Why do you create the Job object on the client side where it will never be
> executed?
>
> My suggestion is to move the job.Job() related code from the client to the
> server and just pass the three strings to the server.
>
> Then the Job object is created on the server side where it will be
> executed. The client can still influence the environment, the argument and
> the cmd.
>
> In case you want that the files for the environment and the commands are
> actually supplied by the client to the server you can handle this by
> supplying a callable (callback function) to the server to access local
> files on the client. (You might want to limit the access to certain files
> or dirs for security reasons).
>
> Some time ago I wrote a numerical client server tool (Monte Carlo
> Simualtion) with rpyc where a client can employ a number of servers to do
> repeated calculations. The code of for the calculations is supplied by the
> client (using xml). The server communcates the calculation results back to
> the client using numpy arrays. At the end the system could handle any
> number of servers with any number of simultanious (different) jobs and
> clients. By definition one client could execute one job at a time.
>
> I think this pattern was similar to yours and it was rather straight
> forward to implement it by using rpyc. In case you use callback from the
> server to the client you need to make sure that a thread is listening for
> incomming rpyc request from the server on the client side while the server
> is working.
>
> Greetings
> Rüdiger
>
>
>
> 2015-09-21 14:43 GMT+02:00 Michael Mann <[email protected] <javascript:>>:
>
>> Simplified code below...
>>
>>
>> *SERVER CODE:*
>>
>>
>> #!/usr/bin/env python
>> import os
>> import util
>> import job
>> import multiprocessing
>> import rpyc
>>
>> class MyClass(rpyc.Service):
>> """ Manages jobs requested on a Node instance"""
>>
>> def exposed_submitJobObj(self,newJob):
>> """Create a new Job using the input arguments"""
>>
>> self.addJobToQueue(newJob)
>>
>>
>> def addJobToQueue(self,newJob):
>> """Adds a new job to the node job execution queue and executes
>> it"""
>>
>> #execute the job in a new thread
>> myJobWorker = multiprocessing.Process(target=self.jobWorker, args
>> =(newJob,))
>> myJobWorker.start()
>>
>>
>> def jobWorker(self,currJob):
>> """Multiprocess method to run job as a parallel process """
>> currJob.execute()
>>
>>
>> def startNode():
>>
>>
>> from rpyc.utils.server import ThreadedServer
>> t = ThreadedServer(NodeManager, port = 18861, protocol_config = {
>> "allow_public_attrs" : True,"allow_pickle":True})
>> t.start()
>>
>>
>> if __name__ == '__main__':
>> startNode()
>>
>>
>>
>> *CLIENT CODE:*
>>
>> #!/usr/bin/env python
>> import os
>> import job
>> import util
>> import rpyc
>>
>> def testJobWrapper():
>>
>> c = rpyc.connect("localhost", 18861, config = {"allow_public_attrs" :
>> True,"allow_pickle":True})
>>
>> winenv='C:\\mydir\\mysubdir\\test.env'
>> winarg=['param1']
>> wincmd = 'C:\\mydir\\mysubdir\\hellow.bat'
>> winJob = job.Job(wincmd,winarg)
>> testJob = job.JobWrapper()
>> testJob.setJobAttributes(winenv,winJob)
>> c.root.exposed_submitJobObj(testJob)
>>
>>
>> if __name__ == '__main__':
>> #testJob()
>> testJobWrapper()
>>
>>
>> --
>>
>> ---
>> You received this message because you are subscribed to the Google Groups
>> "rpyc" group.
>> To unsubscribe from this group and stop receiving emails from it, send an
>> email to [email protected] <javascript:>.
>> For more options, visit https://groups.google.com/d/optout.
>>
>
>
--
---
You received this message because you are subscribed to the Google Groups
"rpyc" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
For more options, visit https://groups.google.com/d/optout.
