KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > objectweb > proactive > core > process > globus > GridJob


1 /*
2  * ################################################################
3  *
4  * ProActive: The Java(TM) library for Parallel, Distributed,
5  * Concurrent computing with Security and Mobility
6  *
7  * Copyright (C) 1997-2002 INRIA/University of Nice-Sophia Antipolis
8  * Contact: proactive-support@inria.fr
9  *
10  * This library is free software; you can redistribute it and/or
11  * modify it under the terms of the GNU Lesser General Public
12  * License as published by the Free Software Foundation; either
13  * version 2.1 of the License, or any later version.
14  *
15  * This library is distributed in the hope that it will be useful,
16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
18  * Lesser General Public License for more details.
19  *
20  * You should have received a copy of the GNU Lesser General Public
21  * License along with this library; if not, write to the Free Software
22  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
23  * USA
24  *
25  * Initial developer(s): The ProActive Team
26  * http://www.inria.fr/oasis/ProActive/contacts.html
27  * Contributor(s):
28  *
29  * ################################################################
30  */
31 package org.objectweb.proactive.core.process.globus;
32
33 import org.apache.log4j.Logger;
34 import org.globus.gram.*;
35
36 import org.globus.io.gass.server.*;
37
38 //import org.globus.security.*;
39
40 import org.globus.util.deactivator.Deactivator;
41
42 import org.gridforum.jgss.ExtendedGSSManager;
43
44 import org.ietf.jgss.GSSCredential JavaDoc;
45
46
47 /**
48  * Java CoG Job submission class
49  */
50 public class GridJob implements GramJobListener{
51     static Logger logger = Logger.getLogger(GridJob.class.getName());
52     private GassServer m_gassServer; // GASS Server: required to get job output
53 private String JavaDoc m_gassURL = null; // URL of the GASS server
54 private GramJob m_job = null; // GRAM JOB to be executed
55 private String JavaDoc m_jobOutput = ""; // job output as string
56 private boolean m_batch = false; // Submission modes:
57 private String JavaDoc m_remoteHost = null; // host where job will run
58 // private JobOutputStream m_stdoutStream = null;
59 // private JobOutputStream m_stderrStream = null;
60 private String JavaDoc m_jobid = null; // Globus job id on the form:
61 int options =
62         org.globus.io.gass.server.GassServer.READ_ENABLE |
63         org.globus.io.gass.server.GassServer.WRITE_ENABLE |
64         org.globus.io.gass.server.GassServer.STDOUT_ENABLE |
65         org.globus.io.gass.server.GassServer.STDERR_ENABLE;
66
67     //https://server.com:39374/15621/1021382777/
68 public GridJob(String JavaDoc Contact, boolean batch) {
69         m_remoteHost = Contact; // remote host
70 m_batch = batch; // submission mode
71 }
72
73     public String JavaDoc GlobusRun(String JavaDoc RSL) {
74         try {
75             // load default Globus proxy. Java CoG kit must be installed
76 // and a user certificate setup properly
77 ExtendedGSSManager manager = (ExtendedGSSManager) ExtendedGSSManager.getInstance();
78             GSSCredential JavaDoc cred = manager.createCredential(GSSCredential.INITIATE_AND_ACCEPT);
79             if (cred == null){
80                 logger.error("credential null");
81             }
82
83             // Start GASS server
84 if (!startGassServer(cred)) {
85                 throw new Exception JavaDoc("Unable to start GASS server.");
86             }
87
88             // setup Job Output listeners
89 initJobOutListeners();
90
91             // Append GASS URL to job String so we can get some output back
92 String JavaDoc newRSL = null;
93
94             // if non-batch, then get some output back
95 // if (!m_batch) {
96 // newRSL = "&" + RSL.substring(0, RSL.indexOf('&')) +
97 // "(rsl_substitution=(GLOBUSRUN_GASS_URL " + m_gassURL +
98 // "))" + RSL.substring(RSL.indexOf('&') + 1, RSL.length())+
99 // "(stdout=$(GLOBUSRUN_GASS_URL)/dev/stdout-5)(stderr=$(GLOBUSRUN_GASS_URL)/dev/sterr-5)";
100 // //newRSL = RSL;
101 // System.out.println(newRSL);
102 // } else {
103 // // format batching RSL so output can be retrieved later on using any GTK commands
104 // newRSL = RSL +
105 // "(stdout=x-gass-cache://$(GLOBUS_GRAM_JOB_CONTACT)stdout anExtraTag)" +
106 // "(stderr=x-gass-cache://$(GLOBUS_GRAM_JOB_CONTACT)stderr anExtraTag)";
107 // }
108
109                             newRSL = "&" + RSL.substring(0, RSL.indexOf('&')) +
110                                 "(rsl_substitution=(GLOBUSRUN_GASS_URL " + m_gassURL +
111                                 "))" + RSL.substring(RSL.indexOf('&') + 1, RSL.length())+
112                                 "(stdout=$(GLOBUSRUN_GASS_URL)/dev/stdout-5)(stderr=$(GLOBUSRUN_GASS_URL)/dev/sterr-5)";
113                             //newRSL = RSL;
114
115                         
116             Gram.ping(m_remoteHost);
117             System.out.println("ping successfull");
118             System.out.println(newRSL);
119             m_job = new GramJob(newRSL);
120
121             // set proxy. CoG kit and user credentials must be installed and set
122 // up properly
123 m_job.setCredentials(cred);
124
125             // if non-batch then listen for output
126 // if (!m_batch) {
127 //m_job.addListener(new GramJobListenerImpl());
128 m_job.addListener(this);
129            // }
130
131             System.out.println("Sending job request to: " + m_remoteHost);
132             m_job.request(m_remoteHost, m_batch, false);
133
134             // Wait for job to complete
135 // if (!m_batch) {
136 // synchronized (this) {
137 // try {
138 // wait();
139 // } catch (Exception e) {
140 // e.printStackTrace();
141 // }
142 // }
143 //} else {
144 // do not wait for job. Return immediately
145 m_jobOutput = "Job sent. url=" + m_job.getIDAsString();
146            // }
147 } catch (Exception JavaDoc ex) {
148             if (m_gassServer != null) {
149                 // unregister from gass server
150 m_gassServer.unregisterJobOutputStream("err-5");
151                 m_gassServer.unregisterJobOutputStream("out-5");
152             }
153
154             m_jobOutput = "Error submitting job: " + ex.getClass() + ":" +
155                 ex.getMessage();
156                 ex.printStackTrace();
157         }
158
159         // cleanup
160 Deactivator.deactivateAll();
161         return m_jobOutput;
162     }
163
164     /**
165      * Start the Globus GASS Server. Used to get the output from the server
166      * back to the client.
167      */
168     private boolean startGassServer(GSSCredential JavaDoc cred) {
169         if (m_gassServer != null) {
170             return true;
171         }
172
173         try {
174             m_gassServer = new GassServer(cred, 0);
175             m_gassServer.setOptions(options);
176             m_gassURL = m_gassServer.getURL();
177             System.out.println("gass server started succesfully "+m_gassURL);
178         } catch (Exception JavaDoc e) {
179             System.err.println("gass server failed to start!");
180             e.printStackTrace();
181
182             return false;
183         }
184
185         m_gassServer.registerDefaultDeactivator();
186
187         return true;
188     }
189
190     /**
191     * Init job out listeners for non-batch mode jobs.
192     */
193     private void initJobOutListeners() throws Exception JavaDoc {
194         // job output vars
195 JobOutputListenerImpl outListener = new JobOutputListenerImpl();
196         JobOutputStream outStream = new JobOutputStream(outListener);
197         m_gassServer.registerJobOutputStream("out-5", outStream);
198         m_gassServer.registerJobOutputStream("err-5", outStream);
199     }
200     
201     public void statusChanged(GramJob job) {
202                 String JavaDoc status = job.getStatusAsString();
203                 System.out.println("status changed "+status);
204
205                 try {
206                     if (job.getStatus() == GramJob.STATUS_ACTIVE) {
207                         // notify waiting thread when job ready
208 m_jobOutput = "Job sent. url=" + job.getIDAsString();
209                         logger.info(m_jobOutput);
210
211                         // if notify enabled return URL as output
212 // synchronized (this) {
213 // notify();
214 // }
215 }
216                 } catch (Exception JavaDoc ex) {
217                     System.out.println("statusChanged Error:" + ex.getMessage());
218                     ex.printStackTrace();
219                 }
220             }
221     
222
223     private class JobOutputListenerImpl implements JobOutputListener {
224         public void outputClosed() {
225             System.out.println("output closed");
226         }
227
228         public void outputChanged(String JavaDoc output) {
229             m_jobOutput += output;
230             System.out.println("output changed: "+m_jobOutput);
231         }
232     }
233 }
234
Popular Tags