jobs.php 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525
  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('Job ID', 'User', 'Job Name', 'Dates', 'Priority', 'Progress', 'Status', 'Action');
  186. $rows = array();
  187. // iterate through the jobs
  188. while ($job = db_fetch_object($jobs)) {
  189. $submit = tripal_jobs_get_submit_date($job);
  190. $start = tripal_jobs_get_start_time($job);
  191. $end = tripal_jobs_get_end_time($job);
  192. $cancel_link = '';
  193. if ($job->start_time == 0 and $job->end_time == 0) {
  194. $cancel_link = "<a href=\"" . url("admin/tripal/tripal_jobs/cancel/" . $job->job_id) . "\">Cancel</a><br />";
  195. }
  196. $rerun_link = "<a href=\"" . url("admin/tripal/tripal_jobs/rerun/" . $job->job_id) . "\">Re-run</a><br />";
  197. $view_link ="<a href=\"" . url("admin/tripal/tripal_jobs/view/" . $job->job_id) . "\">View</a>";
  198. $rows[] = array(
  199. $job->job_id,
  200. $job->username,
  201. $job->job_name,
  202. "Submit Date: $submit<br>Start Time: $start<br>End Time: $end",
  203. $job->priority,
  204. $job->progress . '%',
  205. $job->job_status,
  206. "$cancel_link $rerun_link $view_link",
  207. );
  208. }
  209. // create the report page
  210. $output .= "Waiting jobs are executed first by priority level (the lower the ".
  211. "number the higher the priority) and second by the order they ".
  212. "were entered";
  213. $output .= drupal_get_form('tripal_jobs_report_form');
  214. $output .= theme('table', $header, $rows);
  215. $output .= theme_pager();
  216. return $output;
  217. }
  218. /**
  219. * Returns the start time for a given job
  220. *
  221. * @param $job
  222. * An object describing the job
  223. *
  224. * @return
  225. * The start time of the job if it was already run and either "Cancelled" or "Not Yet Started" otherwise
  226. *
  227. * @ingroup tripal_jobs_api
  228. */
  229. function tripal_jobs_get_start_time($job) {
  230. if ($job->start_time > 0) {
  231. $start = format_date($job->start_time);
  232. }
  233. else {
  234. if (strcmp($job->job_status, 'Cancelled')==0) {
  235. $start = 'Cancelled';
  236. }
  237. else {
  238. $start = 'Not Yet Started';
  239. }
  240. }
  241. return $start;
  242. }
  243. /**
  244. * Returns the end time for a given job
  245. *
  246. * @param $job
  247. * An object describing the job
  248. *
  249. * @return
  250. * The end time of the job if it was already run and empty otherwise
  251. *
  252. * @ingroup tripal_jobs_api
  253. */
  254. function tripal_jobs_get_end_time($job) {
  255. if ($job->end_time > 0) {
  256. $end = format_date($job->end_time);
  257. }
  258. else {
  259. $end = '';
  260. }
  261. return $end;
  262. }
  263. /**
  264. * Returns the date the job was added to the queue
  265. *
  266. * @param $job
  267. * An object describing the job
  268. *
  269. * @return
  270. * The date teh job was submitted
  271. *
  272. * @ingroup tripal_jobs_api
  273. */
  274. function tripal_jobs_get_submit_date($job) {
  275. return format_date($job->submit_date);
  276. }
  277. /**
  278. * A function used to manually launch all queued tripal jobs
  279. *
  280. * @param $do_parallel
  281. * A boolean indicating whether jobs should be attempted to run in parallel
  282. *
  283. * @param $job_id
  284. * To launch a specific job provide the job id. This option should be
  285. * used sparingly as the jobs queue managment system should launch jobs
  286. * based on order and priority. However there are times when a specific
  287. * job needs to be launched and this argument will allow it. Only jobs
  288. * which have not been run previously will run.
  289. *
  290. * @ingroup tripal_jobs_api
  291. */
  292. function tripal_jobs_launch($do_parallel = 0, $job_id = NULL) {
  293. // first check if any jobs are currently running
  294. // if they are, don't continue, we don't want to have
  295. // more than one job script running at a time
  296. if (!$do_parallel and tripal_jobs_check_running()) {
  297. return;
  298. }
  299. // get all jobs that have not started and order them such that
  300. // they are processed in a FIFO manner.
  301. if ($job_id) {
  302. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  303. "WHERE TJ.start_time IS NULL and TJ.end_time IS NULL and TJ.job_id = %d ".
  304. "ORDER BY priority ASC,job_id ASC";
  305. $job_res = db_query($sql,$job_id);
  306. }
  307. else {
  308. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  309. "WHERE TJ.start_time IS NULL and TJ.end_time IS NULL ".
  310. "ORDER BY priority ASC,job_id ASC";
  311. $job_res = db_query($sql);
  312. }
  313. while ($job = db_fetch_object($job_res)) {
  314. // set the start time for this job
  315. $record = new stdClass();
  316. $record->job_id = $job->job_id;
  317. $record->start_time = time();
  318. $record->status = 'Running';
  319. $record->pid = getmypid();
  320. drupal_write_record('tripal_jobs', $record, 'job_id');
  321. // call the function provided in the callback column.
  322. // Add the job_id as the last item in the list of arguments. All
  323. // callback functions should support this argument.
  324. $callback = $job->callback;
  325. $args = split("::", $job->arguments);
  326. $args[] = $job->job_id;
  327. print "Calling: $callback(" . implode(", ", $args) . ")\n";
  328. call_user_func_array($callback, $args);
  329. // set the end time for this job
  330. $record->end_time = time();
  331. $record->status = 'Completed';
  332. $record->progress = '100';
  333. drupal_write_record('tripal_jobs', $record, 'job_id');
  334. // send an email to the user advising that the job has finished
  335. }
  336. }
  337. /**
  338. * Returns a list of running tripal jobs
  339. *
  340. * @return
  341. * and array of objects where each object describes a running job or FALSE if no jobs are running
  342. *
  343. * @ingroup tripal_jobs_api
  344. */
  345. function tripal_jobs_check_running() {
  346. // iterate through each job that has not ended
  347. // and see if it is still running. If it is not
  348. // running but does not have an end_time then
  349. // set the end time and set the status to 'Error'
  350. $sql = "SELECT * FROM {tripal_jobs} TJ ".
  351. "WHERE TJ.end_time IS NULL and NOT TJ.start_time IS NULL ";
  352. $jobs = db_query($sql);
  353. while ($job = db_fetch_object($jobs)) {
  354. $status = `ps --pid=$job->pid --no-header`;
  355. if ($job->pid && $status) {
  356. // the job is still running so let it go
  357. // we return 1 to indicate that a job is running
  358. print "Job is still running (pid $job->pid)\n";
  359. return TRUE;
  360. }
  361. else {
  362. // the job is not running so terminate it
  363. $record = new stdClass();
  364. $record->job_id = $job->job_id;
  365. $record->end_time = time();
  366. $record->status = 'Error';
  367. $record->error_msg = 'Job has terminated unexpectedly.';
  368. drupal_write_record('tripal_jobs', $record, 'job_id');
  369. }
  370. }
  371. // return 1 to indicate that no jobs are currently running.
  372. return FALSE;
  373. }
  374. /**
  375. * Returns the HTML code to display a given job
  376. *
  377. * @param $job_id
  378. * The job_id of the job to display
  379. *
  380. * @return
  381. * The HTML describing the indicated job
  382. * @ingroup tripal_core
  383. */
  384. function tripal_jobs_view($job_id) {
  385. return theme('tripal_core_job_view', $job_id);
  386. }
  387. /**
  388. * Registers variables for the tripal_core_job_view themeing function
  389. *
  390. * @param $variables
  391. * An array containing all variables supplied to this template
  392. *
  393. * @ingroup tripal_core
  394. */
  395. function tripal_core_preprocess_tripal_core_job_view(&$variables) {
  396. // get the job record
  397. $job_id = $variables['job_id'];
  398. $sql =
  399. "SELECT TJ.job_id,TJ.uid,TJ.job_name,TJ.modulename,TJ.progress,
  400. TJ.status as job_status, TJ,submit_date,TJ.start_time,
  401. TJ.end_time,TJ.priority,U.name as username,TJ.arguments,
  402. TJ.callback,TJ.error_msg,TJ.pid
  403. FROM {tripal_jobs} TJ
  404. INNER JOIN users U on TJ.uid = U.uid
  405. WHERE TJ.job_id = %d";
  406. $job = db_fetch_object(db_query($sql, $job_id));
  407. // we do not know what the arguments are for and we want to provide a
  408. // meaningful description to the end-user. So we use a callback function
  409. // deinfed in the module that created the job to describe in an array
  410. // the arguments provided. If the callback fails then just use the
  411. // arguments as they are
  412. $args = preg_split("/::/", $job->arguments);
  413. $arg_hook = $job->modulename . "_job_describe_args";
  414. if (is_callable($arg_hook)) {
  415. $new_args = call_user_func_array($arg_hook, array($job->callback, $args));
  416. if (is_array($new_args) and count($new_args)) {
  417. $job->arguments = $new_args;
  418. }
  419. else {
  420. $job->arguments = $args;
  421. }
  422. }
  423. else {
  424. $job->arguments = $args;
  425. }
  426. // make our start and end times more legible
  427. $job->submit_date = tripal_jobs_get_submit_date($job);
  428. $job->start_time = tripal_jobs_get_start_time($job);
  429. $job->end_time = tripal_jobs_get_end_time($job);
  430. // add the job to the variables that get exported to the template
  431. $variables['job'] = $job;
  432. }
  433. /**
  434. * Set a job to be re-ran (ie: add it back into the job queue)
  435. *
  436. * @param $job_id
  437. * The job_id of the job to be re-ran
  438. *
  439. * @ingroup tripal_jobs_api
  440. */
  441. function tripal_jobs_rerun($job_id, $goto_jobs_page = TRUE) {
  442. global $user;
  443. $sql = "SELECT * FROM {tripal_jobs} WHERE job_id = %d";
  444. $job = db_fetch_object(db_query($sql, $job_id));
  445. $args = explode("::", $job->arguments);
  446. $job_id = tripal_add_job(
  447. $job->job_name,
  448. $job->modulename,
  449. $job->callback,
  450. $args,
  451. $user->uid,
  452. $job->priority);
  453. if ($goto_jobs_page) {
  454. drupal_goto("admin/tripal/tripal_jobs");
  455. }
  456. return $job_id;
  457. }
  458. /**
  459. * Cancel a Tripal Job currently waiting in the job queue
  460. *
  461. * @param $job_id
  462. * The job_id of the job to be cancelled
  463. *
  464. * @ingroup tripal_jobs_api
  465. */
  466. function tripal_jobs_cancel($job_id) {
  467. $sql = "SELECT * FROM {tripal_jobs} WHERE job_id = %d";
  468. $job = db_fetch_object(db_query($sql, $job_id));
  469. // set the end time for this job
  470. if ($job->start_time == 0) {
  471. $record = new stdClass();
  472. $record->job_id = $job->job_id;
  473. $record->end_time = time();
  474. $record->status = 'Cancelled';
  475. $record->progress = '0';
  476. drupal_write_record('tripal_jobs', $record, 'job_id');
  477. drupal_set_message(t("Job #%job_id cancelled", array('%job_id' => $job_id)));
  478. }
  479. else {
  480. drupal_set_message(t("Job %job_id cannot be cancelled. It is in progress or has finished.", array('%job_id' => $job_id)));
  481. }
  482. drupal_goto("admin/tripal/tripal_jobs");
  483. }