jobs.php 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533
  1. <?php
  2. /**
  3. * @file
  4. * Contains functions related to the Tripal Jobs API
  5. *
  6. * @defgroup tripal_jobs_api Core Module Jobs API
  7. * @{
  8. * Tripal offers a job management subsystem for managing tasks that may require an extended period of time for
  9. * completion. Drupal uses a UNIX-based cron job to handle tasks such as checking the availability of updates,
  10. * indexing new nodes for searching, etc. Drupal's cron uses the web interface for launching these tasks, however,
  11. * Tripal provides several administrative tasks that may time out and not complete due to limitations of the web
  12. * server. Examples including syncing of a large number of features between chado and Drupal. To circumvent this,
  13. * as well as provide more fine-grained control and monitoring, Tripal uses a jobs management sub-system built into
  14. * the Tripal Core module. It is anticipated that this functionality will be used for managing analysis jobs provided by
  15. * future tools, with eventual support for distributed computing.
  16. *
  17. * The Tripal jobs management system allows administrators to submit tasks to be performed which can then be
  18. * launched through a UNIX command-line PHP script or cron job. This command-line script can be added to a cron
  19. * entry along-side the Drupal cron entry for automatic, regular launching of Tripal jobs. The order of execution of
  20. * waiting jobs is determined first by priority and second by the order the jobs were entered.
  21. *
  22. * The API functions described below provide a programmatic interface for adding, checking and viewing jobs.
  23. * @}
  24. * @ingroup tripal_api
  25. */
  26. /**
  27. * Adds a job to the Tripal Jbo queue
  28. *
  29. * @param $job_name
  30. * The human readable name for the job
  31. * @param $modulename
  32. * The name of the module adding the job
  33. * @param $callback
  34. * The name of a function to be called when the job is executed
  35. * @param $arguments
  36. * An array of arguements to be passed on to the callback
  37. * @param $uid
  38. * The uid of the user adding the job
  39. * @param $priority
  40. * The priority at which to run the job where the highest priority is 10 and the lowest priority
  41. * is 1. The default priority is 10.
  42. *
  43. * @return
  44. * The job_id of the registered job
  45. *
  46. * Example usage:
  47. * @code
  48. * $args = array($dfile, $organism_id, $type, $library_id, $re_name, $re_uname,
  49. * $re_accession, $db_id, $rel_type, $re_subject, $parent_type, $method,
  50. * $user->uid, $analysis_id, $match_type);
  51. *
  52. * tripal_add_job("Import FASTA file: $dfile", 'tripal_feature',
  53. * 'tripal_feature_load_fasta', $args, $user->uid);
  54. * @endcode
  55. * The code above is copied from the tripal_feature/fasta_loader.php file. The
  56. * snipped first builds an array of arguments that will then be passed to the
  57. * tripal_add_job function. The number of arguments provided in the $arguments
  58. * variable should match the argument set for the callback function provided
  59. * as the third argument.
  60. *
  61. * @ingroup tripal_jobs_api
  62. */
  63. function tripal_add_job($job_name, $modulename, $callback, $arguments, $uid, $priority = 10) {
  64. // convert the arguments into a string for storage in the database
  65. $args = implode("::", $arguments);
  66. $record = new stdClass();
  67. $record->job_name = $job_name;
  68. $record->modulename = $modulename;
  69. $record->callback = $callback;
  70. $record->status = 'Waiting';
  71. $record->submit_date = time();
  72. $record->uid = $uid;
  73. $record->priority = $priority; # the lower the number the higher the priority
  74. if ($args) {
  75. $record->arguments = $args;
  76. }
  77. if (drupal_write_record('tripal_jobs', $record)) {
  78. $jobs_url = url("admin/tripal/tripal_jobs");
  79. drupal_set_message(t("Job '%job_name' submitted. Check the <a href='!jobs_url'>jobs page</a> for status", array('%job_name' => $job_name, '!jobs_url' => $jobs_url)));
  80. }
  81. else {
  82. drupal_set_message(t("Failed to add job %job_name.", array('%job_name' => $job_name)), 'error');
  83. }
  84. return $record->job_id;
  85. }
  86. /**
  87. * An internal function for setting the progress for a current job
  88. *
  89. * @param $job_id
  90. * The job_id to set the progress for
  91. * @param $percentage
  92. * The progress to set the job to
  93. *
  94. * @return
  95. * True on success and False otherwise
  96. *
  97. * @ingroup tripal_core
  98. */
  99. function tripal_job_set_progress($job_id, $percentage) {
  100. if (preg_match("/^(\d+|100)$/", $percentage)) {
  101. $record = new stdClass();
  102. $record->job_id = $job_id;
  103. $record->progress = $percentage;
  104. if (drupal_write_record('tripal_jobs', $record, 'job_id')) {
  105. return TRUE;
  106. }
  107. }
  108. return FALSE;
  109. }
  110. /**
  111. * Returns a list of jobs associated with the given module
  112. *
  113. * @param $modulename
  114. * The module to return a list of jobs for
  115. *
  116. * @return
  117. * An array of objects where each object describes a tripal job
  118. *
  119. * @ingroup tripal_jobs_api
  120. */
  121. function tripal_get_module_active_jobs($modulename) {
  122. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  123. "WHERE TJ.end_time IS NULL and TJ.modulename = '%s' ";
  124. return db_fetch_object(db_query($sql, $modulename));
  125. }
  126. /**
  127. *
  128. * @ingroup tripal_core
  129. */
  130. function tripal_jobs_report_form($form, &$form_state = NULL) {
  131. $form = array();
  132. // set the default values
  133. $default_status = $form_state['values']['job_status'];
  134. if (!$default_status) {
  135. $default_status = $_SESSION['tripal_job_status_filter'];
  136. }
  137. $form['job_status'] = array(
  138. '#type' => 'select',
  139. '#title' => t('Filter by Job Status'),
  140. '#default_value' => $default_status,
  141. '#options' => array(
  142. 0 => 'All Jobs',
  143. 'Running' => 'Running',
  144. 'Completed' => 'Completed',
  145. 'Cancelled' => 'Cancelled',
  146. 'Error' => 'Error',
  147. ),
  148. );
  149. $form['submit'] = array(
  150. '#type' => 'submit',
  151. '#value' => t('Filter'),
  152. );
  153. return $form;
  154. }
  155. /**
  156. *
  157. * @ingroup tripal_core
  158. */
  159. function tripal_jobs_report_form_submit($form, &$form_state = NULL) {
  160. $job_status = $form_state['values']['job_status'];
  161. $_SESSION['tripal_job_status_filter'] = $job_status;
  162. }
  163. /**
  164. * Returns the Tripal Job Report
  165. *
  166. * @return
  167. * The HTML to be rendered which describes the job report
  168. *
  169. * @ingroup tripal_core
  170. */
  171. function tripal_jobs_report() {
  172. $jobs_status_filter = $_SESSION['tripal_job_status_filter'];
  173. $sql = "
  174. SELECT
  175. TJ.job_id,TJ.uid,TJ.job_name,TJ.modulename,TJ.progress,
  176. TJ.status as job_status, TJ,submit_date,TJ.start_time,
  177. TJ.end_time,TJ.priority,U.name as username
  178. FROM {tripal_jobs} TJ
  179. INNER JOIN {users} U on TJ.uid = U.uid ";
  180. if ($jobs_status_filter) {
  181. $sql .= "WHERE TJ.status = '%s' ";
  182. }
  183. $sql .= "ORDER BY job_id DESC";
  184. $jobs = pager_query($sql, 25, 0, "SELECT count(*) FROM ($sql) as t1", $jobs_status_filter);
  185. $header = array(
  186. 'Job ID',
  187. 'User',
  188. 'Job Name',
  189. array('data' => 'Dates', 'style'=> "white-space: nowrap"),
  190. 'Priority',
  191. 'Progress',
  192. 'Status',
  193. 'Action');
  194. $rows = array();
  195. // iterate through the jobs
  196. while ($job = db_fetch_object($jobs)) {
  197. $submit = tripal_jobs_get_submit_date($job);
  198. $start = tripal_jobs_get_start_time($job);
  199. $end = tripal_jobs_get_end_time($job);
  200. $cancel_link = '';
  201. if ($job->start_time == 0 and $job->end_time == 0) {
  202. $cancel_link = "<a href=\"" . url("admin/tripal/tripal_jobs/cancel/" . $job->job_id) . "\">Cancel</a><br />";
  203. }
  204. $rerun_link = "<a href=\"" . url("admin/tripal/tripal_jobs/rerun/" . $job->job_id) . "\">Re-run</a><br />";
  205. $view_link ="<a href=\"" . url("admin/tripal/tripal_jobs/view/" . $job->job_id) . "\">View</a>";
  206. $rows[] = array(
  207. $job->job_id,
  208. $job->username,
  209. $job->job_name,
  210. "Submit Date: $submit<br>Start Time: $start<br>End Time: $end",
  211. $job->priority,
  212. $job->progress . '%',
  213. $job->job_status,
  214. "$cancel_link $rerun_link $view_link",
  215. );
  216. }
  217. // create the report page
  218. $output .= "Waiting jobs are executed first by priority level (the lower the ".
  219. "number the higher the priority) and second by the order they ".
  220. "were entered";
  221. $output .= drupal_get_form('tripal_jobs_report_form');
  222. $output .= theme('table', $header, $rows);
  223. $output .= theme_pager();
  224. return $output;
  225. }
  226. /**
  227. * Returns the start time for a given job
  228. *
  229. * @param $job
  230. * An object describing the job
  231. *
  232. * @return
  233. * The start time of the job if it was already run and either "Cancelled" or "Not Yet Started" otherwise
  234. *
  235. * @ingroup tripal_jobs_api
  236. */
  237. function tripal_jobs_get_start_time($job) {
  238. if ($job->start_time > 0) {
  239. $start = format_date($job->start_time);
  240. }
  241. else {
  242. if (strcmp($job->job_status, 'Cancelled')==0) {
  243. $start = 'Cancelled';
  244. }
  245. else {
  246. $start = 'Not Yet Started';
  247. }
  248. }
  249. return $start;
  250. }
  251. /**
  252. * Returns the end time for a given job
  253. *
  254. * @param $job
  255. * An object describing the job
  256. *
  257. * @return
  258. * The end time of the job if it was already run and empty otherwise
  259. *
  260. * @ingroup tripal_jobs_api
  261. */
  262. function tripal_jobs_get_end_time($job) {
  263. if ($job->end_time > 0) {
  264. $end = format_date($job->end_time);
  265. }
  266. else {
  267. $end = '';
  268. }
  269. return $end;
  270. }
  271. /**
  272. * Returns the date the job was added to the queue
  273. *
  274. * @param $job
  275. * An object describing the job
  276. *
  277. * @return
  278. * The date teh job was submitted
  279. *
  280. * @ingroup tripal_jobs_api
  281. */
  282. function tripal_jobs_get_submit_date($job) {
  283. return format_date($job->submit_date);
  284. }
  285. /**
  286. * A function used to manually launch all queued tripal jobs
  287. *
  288. * @param $do_parallel
  289. * A boolean indicating whether jobs should be attempted to run in parallel
  290. *
  291. * @param $job_id
  292. * To launch a specific job provide the job id. This option should be
  293. * used sparingly as the jobs queue managment system should launch jobs
  294. * based on order and priority. However there are times when a specific
  295. * job needs to be launched and this argument will allow it. Only jobs
  296. * which have not been run previously will run.
  297. *
  298. * @ingroup tripal_jobs_api
  299. */
  300. function tripal_jobs_launch($do_parallel = 0, $job_id = NULL) {
  301. // first check if any jobs are currently running
  302. // if they are, don't continue, we don't want to have
  303. // more than one job script running at a time
  304. if (!$do_parallel and tripal_jobs_check_running()) {
  305. return;
  306. }
  307. // get all jobs that have not started and order them such that
  308. // they are processed in a FIFO manner.
  309. if ($job_id) {
  310. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  311. "WHERE TJ.start_time IS NULL and TJ.end_time IS NULL and TJ.job_id = %d ".
  312. "ORDER BY priority ASC,job_id ASC";
  313. $job_res = db_query($sql,$job_id);
  314. }
  315. else {
  316. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  317. "WHERE TJ.start_time IS NULL and TJ.end_time IS NULL ".
  318. "ORDER BY priority ASC,job_id ASC";
  319. $job_res = db_query($sql);
  320. }
  321. while ($job = db_fetch_object($job_res)) {
  322. // set the start time for this job
  323. $record = new stdClass();
  324. $record->job_id = $job->job_id;
  325. $record->start_time = time();
  326. $record->status = 'Running';
  327. $record->pid = getmypid();
  328. drupal_write_record('tripal_jobs', $record, 'job_id');
  329. // call the function provided in the callback column.
  330. // Add the job_id as the last item in the list of arguments. All
  331. // callback functions should support this argument.
  332. $callback = $job->callback;
  333. $args = split("::", $job->arguments);
  334. $args[] = $job->job_id;
  335. print "Calling: $callback(" . implode(", ", $args) . ")\n";
  336. call_user_func_array($callback, $args);
  337. // set the end time for this job
  338. $record->end_time = time();
  339. $record->status = 'Completed';
  340. $record->progress = '100';
  341. drupal_write_record('tripal_jobs', $record, 'job_id');
  342. // send an email to the user advising that the job has finished
  343. }
  344. }
  345. /**
  346. * Returns a list of running tripal jobs
  347. *
  348. * @return
  349. * and array of objects where each object describes a running job or FALSE if no jobs are running
  350. *
  351. * @ingroup tripal_jobs_api
  352. */
  353. function tripal_jobs_check_running() {
  354. // iterate through each job that has not ended
  355. // and see if it is still running. If it is not
  356. // running but does not have an end_time then
  357. // set the end time and set the status to 'Error'
  358. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  359. "WHERE TJ.end_time IS NULL and NOT TJ.start_time IS NULL ";
  360. $jobs = db_query($sql);
  361. while ($job = db_fetch_object($jobs)) {
  362. $status = `ps --pid=$job->pid --no-header`;
  363. if ($job->pid && $status) {
  364. // the job is still running so let it go
  365. // we return 1 to indicate that a job is running
  366. print "Job is still running (pid $job->pid)\n";
  367. return TRUE;
  368. }
  369. else {
  370. // the job is not running so terminate it
  371. $record = new stdClass();
  372. $record->job_id = $job->job_id;
  373. $record->end_time = time();
  374. $record->status = 'Error';
  375. $record->error_msg = 'Job has terminated unexpectedly.';
  376. drupal_write_record('tripal_jobs', $record, 'job_id');
  377. }
  378. }
  379. // return 1 to indicate that no jobs are currently running.
  380. return FALSE;
  381. }
  382. /**
  383. * Returns the HTML code to display a given job
  384. *
  385. * @param $job_id
  386. * The job_id of the job to display
  387. *
  388. * @return
  389. * The HTML describing the indicated job
  390. * @ingroup tripal_core
  391. */
  392. function tripal_jobs_view($job_id) {
  393. return theme('tripal_core_job_view', $job_id);
  394. }
  395. /**
  396. * Registers variables for the tripal_core_job_view themeing function
  397. *
  398. * @param $variables
  399. * An array containing all variables supplied to this template
  400. *
  401. * @ingroup tripal_core
  402. */
  403. function tripal_core_preprocess_tripal_core_job_view(&$variables) {
  404. // get the job record
  405. $job_id = $variables['job_id'];
  406. $sql =
  407. "SELECT TJ.job_id,TJ.uid,TJ.job_name,TJ.modulename,TJ.progress,
  408. TJ.status as job_status, TJ,submit_date,TJ.start_time,
  409. TJ.end_time,TJ.priority,U.name as username,TJ.arguments,
  410. TJ.callback,TJ.error_msg,TJ.pid
  411. FROM {tripal_jobs} TJ
  412. INNER JOIN users U on TJ.uid = U.uid
  413. WHERE TJ.job_id = %d";
  414. $job = db_fetch_object(db_query($sql, $job_id));
  415. // we do not know what the arguments are for and we want to provide a
  416. // meaningful description to the end-user. So we use a callback function
  417. // deinfed in the module that created the job to describe in an array
  418. // the arguments provided. If the callback fails then just use the
  419. // arguments as they are
  420. $args = preg_split("/::/", $job->arguments);
  421. $arg_hook = $job->modulename . "_job_describe_args";
  422. if (is_callable($arg_hook)) {
  423. $new_args = call_user_func_array($arg_hook, array($job->callback, $args));
  424. if (is_array($new_args) and count($new_args)) {
  425. $job->arguments = $new_args;
  426. }
  427. else {
  428. $job->arguments = $args;
  429. }
  430. }
  431. else {
  432. $job->arguments = $args;
  433. }
  434. // make our start and end times more legible
  435. $job->submit_date = tripal_jobs_get_submit_date($job);
  436. $job->start_time = tripal_jobs_get_start_time($job);
  437. $job->end_time = tripal_jobs_get_end_time($job);
  438. // add the job to the variables that get exported to the template
  439. $variables['job'] = $job;
  440. }
  441. /**
  442. * Set a job to be re-ran (ie: add it back into the job queue)
  443. *
  444. * @param $job_id
  445. * The job_id of the job to be re-ran
  446. *
  447. * @ingroup tripal_jobs_api
  448. */
  449. function tripal_jobs_rerun($job_id, $goto_jobs_page = TRUE) {
  450. global $user;
  451. $sql = "SELECT * FROM {tripal_jobs} WHERE job_id = %d";
  452. $job = db_fetch_object(db_query($sql, $job_id));
  453. $args = explode("::", $job->arguments);
  454. $job_id = tripal_add_job(
  455. $job->job_name,
  456. $job->modulename,
  457. $job->callback,
  458. $args,
  459. $user->uid,
  460. $job->priority);
  461. if ($goto_jobs_page) {
  462. drupal_goto("admin/tripal/tripal_jobs");
  463. }
  464. return $job_id;
  465. }
  466. /**
  467. * Cancel a Tripal Job currently waiting in the job queue
  468. *
  469. * @param $job_id
  470. * The job_id of the job to be cancelled
  471. *
  472. * @ingroup tripal_jobs_api
  473. */
  474. function tripal_jobs_cancel($job_id) {
  475. $sql = "SELECT * FROM {tripal_jobs} WHERE job_id = %d";
  476. $job = db_fetch_object(db_query($sql, $job_id));
  477. // set the end time for this job
  478. if ($job->start_time == 0) {
  479. $record = new stdClass();
  480. $record->job_id = $job->job_id;
  481. $record->end_time = time();
  482. $record->status = 'Cancelled';
  483. $record->progress = '0';
  484. drupal_write_record('tripal_jobs', $record, 'job_id');
  485. drupal_set_message(t("Job #%job_id cancelled", array('%job_id' => $job_id)));
  486. }
  487. else {
  488. drupal_set_message(t("Job %job_id cannot be cancelled. It is in progress or has finished.", array('%job_id' => $job_id)));
  489. }
  490. drupal_goto("admin/tripal/tripal_jobs");
  491. }