forester/archive/perl/queue.pm

   1 package queue;
   2
   3 # Process queueing
   4 # SRE, Wed Sep  2 14:37:14 1998
   5 # CVS $Id: queue.pm,v 1.1.1.1 2005/03/22 08:35:51 cmzmasek Exp $
   6 # Master copy: see src/queue (CVS controlled, separate from pfamserver)
   7 #
   8 # Written for Pfam web server; suited for queuing any set of commands.
   9 #
  10 # API:
  11 #
  12 # $mypid = $$;
  13 # $delay_in_seconds = 2;
  14 #
  15 # $nqueued = &queue::CheckQueue("pfamqueue", "username", "/tmp");
  16 # print ("There are $nqueued jobs ahead of you in line\n");
  17 # &queue::WaitInQueue("pfamqueue", "username", "/tmp", $mypid, $delay_in_seconds);
  18 # print ("Our turn! Working...\n");
  19 # (do stuff)
  20 # &queue::RemoveFromQueue("pfamqueue", "username", "/tmp", $mypid);
  21 #
  22 # queuedir is a directory where the script has write permissions;
  23 # typically a tmp directory of some sort.
  24 #
  25
  26
  27 ################################################################
  28 # PFAMSERVER - The Washington University/St. Louis Pfam web server
  29 # Copyright (C) 1995-1999 Washington University School of Medicine
  30 # Copyright (C) 1995-1999 Sanger Centre/Genome Research Ltd.
  31 # Copyright (C) 1998-1999 Karolinska Institutet Center for Genomics Research
  32 # All Rights Reserved
  33 #
  34 #     This source code is distributed under the terms of the
  35 #     GNU General Public License. See the files COPYRIGHT and LICENSE
  36 #     for details.
  37 #
  38 ################################################################
  39 # RCS $Id: queue.pm,v 1.1.1.1 2005/03/22 08:35:51 cmzmasek Exp $
  40
  41
  42 # WaitInQueue() - add a process id to a queue, wait for turn
  43 #
  44 # Arguments: queue    - name of queue (prefix of queue stamp)
  45 #            username - name of user (middle part of queue stamp)
  46 #            queuedir - directory to keep queue stamps in
  47 #            mypid    - our process id
  48 #            delay    - number of seconds between checking queue status
  49 #
  50 # Note: When it checks the queue, if a stamp is present that
  51 #       doesn't seem to correspond to a running process (ps -a),
  52 #       it deletes the stamp. This protects against crashed processes
  53 #       freezing all subsequent jobs.
  54 #
  55 # example: &WaitInQueue("pfamqueue", "/tmp", $mypid, 2);
  56 #
  57 # Returns 1 on success, 0 on failure.
  58 #
  59 # NOTE: You may have to set the ps command in WaitInQueue.
  60 #       It must return all running processes.
  61 #
  62 sub WaitInQueue
  63 {
  64     local($queue, $username, $queuedir, $mypid, $delay) = @_;
  65     local(@newqueue, @queuelist, %mark);
  66     local(*STAMP, *QUEUEDIR);
  67     local(%is_running);
  68     local(@output, $line, $pid, $waiting);
  69
  70                                 # get list of other guys who are working
  71     opendir(QUEUEDIR, $queuedir);
  72     @queuelist = grep(/$queue\.\S*\.\d+/, readdir(QUEUEDIR));
  73     closedir(QUEUEDIR);
  74                                 # make stamp for our pid
  75     if ($username eq "") { $username = "unknown"; }
  76     open(STAMP, ">$queuedir/$queue.$username.$mypid") || return 0;
  77     close(STAMP);
  78                                 # wait for our turn
  79     while (1)
  80     {
  81         if ($#queuelist == -1) { last; } # nobody ahead of us; our turn!
  82         sleep($delay);
  83                                 # get list of running processes
  84         %is_running = 0;
  85         @output = split(/^/, `ps -ax`);
  86         foreach $line (@output) {
  87             $line =~ /\s*(\d+)/;
  88             $is_running{$1} = 1;
  89         }
  90                                 # verify that the guys we're waiting for
  91                                 # are still running, and haven't crashed.
  92                                 # if they have, reap their stamps, and their
  93                                 # tmp files.
  94         foreach $waiting (@queuelist) {
  95             ($name, $pid) = ($waiting =~ /$queue\.(\S*)\.(\d+)/);
  96             if (! $is_running{$pid}) { unlink "$queuedir/$queue.$name.$pid"; }
  97         }
  98
  99         # get new list of queued jobs ahead of us.
 100         # ignore guys who came in after we  grabbed our initial queue list;
 101         # they're waiting for *us*. The crazed greps are the Perl-y
 102         # way of computing an intersection between two arrays.
 103         #
 104         opendir(QUEUEDIR, $queuedir);
 105         @newqueue = grep(/$queue\.\S*\.\d+/, readdir(QUEUEDIR));
 106         closedir(QUEUEDIR);
 107         %mark = 0;
 108         grep($mark{$_}++,@queuelist);
 109         @queuelist = grep($mark{$_},@newqueue);
 110     }
 111
 112     1;                          # time to run!
 113 }
 114
 115
 116 # CheckQueue() - return total number of processes working, other than us
 117 #                and the total that this particular username is running.
 118 #
 119 # Arguments: queue, username, queuedir
 120 #
 121 sub CheckQueue
 122 {
 123     local($queue, $username, $queuedir) = @_;
 124     local(*QUEUEDIR, @allqueue, $nall, $nuser);
 125
 126     opendir(QUEUEDIR, $queuedir);
 127     @allqueue = grep(/$queue\.\S*\.\d+/, readdir(QUEUEDIR));
 128     closedir(QUEUEDIR);
 129
 130     if ($username eq "") {$username = "unknown"; }
 131     $nall = $nuser = 0;
 132     foreach $waiting (@allqueue) {
 133         ($name, $pid) = ($waiting =~ /$queue\.(\S*)\.(\d+)/);
 134         $nall++;
 135         if ($name eq $username) { $nuser++; }
 136     }
 137     return ($nall, $nuser);
 138 }
 139
 140
 141 # RemoveFromQueue() - remove a pid from a queue
 142 #
 143 sub RemoveFromQueue
 144 {
 145     local($queue, $username, $queuedir, $pid) = @_;
 146     if ($username eq "") {$username = "unknown"; }
 147     unlink "$queuedir/$queue.$username.$pid";
 148 }
 149
 150 1;