This is Executive.m in view mode; [Download] [Up]
/*------------------------------------------------------------------------------
Executive.m Copyright (c) 1990, Marc A. Davidson
SYNOPSIS
Handles executing commands on files either through system() or pipe().
Using the execute:async: method or the pipe:async: method you can execute a
command asynchronous to your main thread.
To handle asynchronous commands the object uses a DPSTimedEntry. The
timed entry is created when the first asynchronous command is issued, and
it persists until all commands that have been issued have completed.
Synchronization between the threads and the timed entry is handled by
two queues: running and done. When a command is first forked, the
description is placed in a command_d struct and put into the running
queue. The timed entry is then created which constantly monitors the
done queue. The command_d is removed from the running queue and placed
in the done queue by the forked thread when the command completes. The
timed entry removes the entry from the done queue, decrements the counter
of currently running commands. If it removes the last one (the counter
is decremented to 0), then it removes itself from the timed entry
scheduler. The timed entry is also used to send lines of output from an
asynchronous pipe:async: command.
Some notes about the implementation:
First, the timed entry mechanism is cumbersome, but there is no other way
to synchronize the main thread (the caller's thread) and the command thread
transparently.
Second, the use of List objects in this object is a little unorthodox.
Since the List object is designed for lists of objects, using it for storing
structs is not strictly correct. The reason for this behavior is that a
supplementary object that holds the same information as the command_d would
need to be created, resulting in two objects instead of one. Besides, the
List object doesn't seem to care what is being manipulated (as long as you
don't use the freeObjects method!), just as long as there are pointers
involved (NeXT will probably give me the hairy eyeball for this).
REVISIONS
Date Who Modification
24-Sept-90 MAD Created (simple synchronous routines)
5-Oct-90 MAD Added cthread routines and queues for executing
asynchronous commands.
9-Oct-90 MAD Added asynchronous capability for pipe: methods as well.
After finishing, I fell asleep on the keyboarddddddddd.
------------------------------------------------------------------------------*/
# import <stdlib.h>
# import <ctype.h>
# import <string.h>
# import <libc.h>
# import <errno.h>
# import <appkit/Application.h>
# import <appkit/Panel.h>
# import <dpsclient/dpsNeXT.h>
# import <objc/List.h>
# import "Executive.h"
/*
* Type codes for command list entry type
*/
# define COMMAND_EXECUTE 0
# define COMMAND_PIPE 1
# define COMMAND_PIPERESULT 2
/*
* Default period at which our main thread synchronizing timed entry executes.
*/
# define DEFAULT_NOTIFY_PERIOD 0.25
/*
* this is the struct that the asynchronous routines depend upon. They are entered
* sent into the running queue, removed and sent into the done queue, and removed
* from the done queue by the DPSTimedEntry. Piped commands send their lines into
* the done queue for communication with the main thread.
*/
typedef struct command_t
{ int commandId; /* unique identifier for command */
int type; /* COMMAND_PIPE, COMMAND_PIPERESULT, COMMAND_EXECUTE */
id self; /* for communication with main thread */
char *command; /* command that was issued */
char *environs; /* environment string that was given */
int result; /* result of system() call (0 in case of pipe) */
char *line; /* the line that was read by pipe */
id to; /* to whom to send this line... */
SEL action; /* ...and method to use */
} command_d;
/*
* newString allocates a new string and returns it, or returns NULL if NULL was
* passed in.
*/
static char *newString(CSTR s)
{ char *ns = NULL;
if (s != NULL)
{ ns = (char *)malloc(strlen(s)+1);
strcpy(ns,s);
}
return(ns);
}
/*
* methods to implement a "cheap object"
*/
static command_d *newCommand(int commandId,int type,CSTR command,CSTR environs,id self)
{ command_d *com;
com = (command_d *) calloc(1,sizeof(command_d));
com->type = type;
com->commandId = commandId;
com->command = newString(command);
com->environs = newString(environs);
com->self = self;
com->result = 0;
return(com);
}
/*
* new Execute command structure
*/
static command_d *newECommand(int commandId,CSTR command,CSTR environs,id self)
{ command_d *nc;
nc = newCommand(commandId,COMMAND_EXECUTE,command,environs,self);
return(nc);
}
/*
* new Pipe command structure
*/
static command_d *newPCommand(int commandId,CSTR command,CSTR environs,id self,
id to,SEL action)
{ command_d *nc;
nc = newCommand(commandId,COMMAND_PIPE,command,environs,self);
nc->to = to;
nc->action = action;
return(nc);
}
/*
* frees any command object
*/
static void freeCommand(command_d *com)
{
if (com->command != NULL)
free(com->command);
if (com->environs != NULL)
free(com->environs);
if ((com->type == COMMAND_PIPERESULT) && (com->line != NULL))
free(com->line); /* free the allocated line */
free(com);
}
static void makeCmdLine(char *cmdLine,CSTR command,CSTR environs)
{
if (environs != NULL)
sprintf(cmdLine,"%s;%s",environs,command);
else
sprintf(cmdLine,"%s",command);
}
static int doExecute(CSTR command,CSTR environs)
{ char cmdNull[MAXPATHLEN];
makeCmdLine(cmdNull,command,environs);
strcat(cmdNull," >/dev/null 2>/dev/null");
return(system(cmdNull));
}
/*
* doPipe works either synchronously or asynchronously.
* Note that the line allocated in each asynchronous iteration is deallocated
* when the to object has been notified via the DPSTimedEntry.
*/
static void doPipe(CSTR command,CSTR environs,BOOL async,id self,int commandId,
id to,SEL aSelector)
{ FILE *fd;
char theLine[MAXPATHLEN],
cmdLine[MAXPATHLEN],
*cp;
extern int pclose(FILE *fd);
makeCmdLine(cmdLine,command,environs);
if ((fd = popen(cmdLine,"r")) != NULL)
{ while (fgets(theLine,MAXPATHLEN,fd) != NULL)
{ if (async)
{ command_d *nc;
nc = newPCommand(commandId,NULL,NULL,self,to,aSelector);
nc->type = COMMAND_PIPERESULT;
nc->line = newString(theLine);
mutex_lock(((Executive *)self)->doneLock);
[((Executive *)self)->done insertObject:(id)nc at:0];
mutex_unlock(((Executive *)self)->doneLock);
cthread_yield();
}
else
[to perform:aSelector with:(id)theLine];
}
pclose(fd);
}
}
@implementation Executive
+ new
{
self = [super new];
curCmdId = 0;
numExecuting = 0;
period = DEFAULT_NOTIFY_PERIOD;
runningLock = mutex_alloc();
doneLock = mutex_alloc();
running = [List new];
done = [List new];
return(self);
}
+ newPeriod:(double)p
{
self = [Executive new];
period = p;
return(self);
}
- free
{
mutex_free(runningLock);
mutex_free(doneLock);
[running free];
[done free];
return([super free]);
}
/*---------------------------------< OUTLET METHODS >--------------------------------*/
- target { return(target); }
- setTarget:anObject { target = anObject; return(self); }
- (SEL) action { return(action); }
- setAction:(SEL)aSelector { action = aSelector; return(self); }
- setPeriod:(double)p { period = p; return(self); }
- (double)period { return(period); }
/*-------------------------------< OVERRIDDEN METHODS >------------------------------*/
/*----------------------------------< OTHER METHODS >--------------------------------*/
/*
* doNotify is called by the timed entry glue function to send messages to the main
* thread in an synchronous manner. For command types it decrements the executing
* counter and removes the timed entry if all commands have completed.
* Notification of a done command is sent to the target when either a pipe command
* or execute command is completed. Pipe results are sent to the given object and
* selector.
*/
- doNotify:(DPSTimedEntry)entry
{ command_d *com;
if ([done count] != 0)
{ mutex_lock(doneLock);
com = (command_d *)[done removeLastObject];
mutex_unlock(doneLock);
switch(com->type)
{ case COMMAND_PIPE:
case COMMAND_EXECUTE:
[target perform:action with:(id)com->commandId with:(id)com->result];
if (--numExecuting == 0)
DPSRemoveTimedEntry(entry);
break;
case COMMAND_PIPERESULT:
[com->to perform:com->action with:(id)com->commandId with:(id)com->line];
break;
default:
fprintf(stderr,"Executive: Retrieved unexpected command type:"
" %d\n",com->type);
[NXApp terminate:self];
break;
}
freeCommand(com);
}
return(self);
}
/*
* commandDone performs the appropriate action when a command is done: removes it
* from the running queue and places it in the done queue for removal by the
* DPSTimedEntry. Since a pipe may be in progress and there may be many lines
* queued for sending, this will insert the program at the very beginning of
* the done queue, thus implementing a priority queue.
*/
- commandDone:(command_d *)com
{
mutex_lock(runningLock);
mutex_lock(doneLock);
[running removeObject:(id)com];
if (com->type == COMMAND_EXECUTE)
[done addObject:(id)com];
else
[done insertObject:(id)com at:0];
mutex_unlock(doneLock);
mutex_unlock(runningLock);
return(self);
}
- (int)doAsyncExecute:(command_d *)com
{ int err;
err = doExecute(com->command,com->environs);
[self commandDone:com];
return(err);
}
/*
* this is a wrapper for the thread that runs the asynchronous command, then
* exits.
*/
void asyncExecute(command_d *com)
{
com->result = [com->self doAsyncExecute:com];
}
- doAsyncPipe:(command_d *)com
{
doPipe(com->command,com->environs,YES,self,com->commandId,com->to,com->action);
[self commandDone:com];
return(self);
}
/*
* this is a wrapper for the thread that runs the asynchronous pipe command, then
* exits.
*/
void asyncPipe(command_d *com)
{
[com->self doAsyncPipe:com];
}
/*
* this is a timed entry that will notify the target of the Executive when the
* asynchronous command identified by a unique number has completed.
*/
static void notify(DPSTimedEntry entry, double now,id self)
{
[self doNotify:entry];
}
/*
* addCommandToRunning adds the given command structure to the running queue,
* performing appropriate mutex_lock's and incrementing the executing count
* to reflect the add.
*/
- addCommandToRunning:(command_d *)com
{
mutex_lock(runningLock);
[running addObject:(id)com];
mutex_unlock(runningLock);
if (++numExecuting == 1)
DPSAddTimedEntry(period,(DPSTimedEntryProc) notify,
(void *)self,NX_BASETHRESHOLD);
return(self);
}
- (int)execute:(CSTR)command
{
return([self execute:command environs:NULL async:NO]);
}
- (int) execute:(CSTR)command async:(BOOL)async
{
return([self execute:command environs:NULL async:async]);
}
- (int) execute:(CSTR)command environs:(CSTR)environs async:(BOOL)async
{ command_d *nc;
if (!async)
return(doExecute(command,environs));
nc = newECommand(++curCmdId,command,environs,self);
cthread_detach(cthread_fork(asyncExecute,nc));
[self addCommandToRunning:nc];
return(nc->commandId);
}
- (int)pipe:(CSTR)command to:anObject :(SEL)aSelector async:(BOOL)async
{
return([self pipe:command environs:NULL to:anObject :aSelector async:async]);
}
- (int)pipe:(CSTR)command environs:(CSTR)environs to:anObject :(SEL)aSelector
async:(BOOL)async
{ command_d *nc;
if (!async)
{ doPipe(command,environs,async,nil,0,anObject,aSelector);
return(0);
}
nc = newPCommand(++curCmdId,command,environs,self,anObject,aSelector);
[self addCommandToRunning:nc];
cthread_detach(cthread_fork(asyncPipe,nc));
return(nc->commandId);
}
- showError:(int)err
{
return([self showError:err while:"executing a command" on:"on a file"
using:"File"]);
}
- showError:(int)err while:(CSTR)doingWhat
{
return([self showError:err while:doingWhat on:"a file" using:"File"]);
}
- showError:(int)err while:(CSTR)doingWhat on:(CSTR)fname
{
return([self showError:err while:doingWhat on:fname using:"File"]);
}
- showError:(int)err while:(CSTR)doingWhat on:(CSTR)fname using:(CSTR)prog
{ char err_str1[MAXPATHLEN],
err_str2[MAXPATHLEN];
sprintf(err_str1,"%s error",prog);
sprintf(err_str2,"Error while %s %s (error code %d)",doingWhat,fname,err);
NXRunAlertPanel(err_str1,err_str2,"Ok",NULL,NULL);
return(self);
}
- showFError:(CSTR)fname
{
NXRunAlertPanel("File error","Can't access %s (%s).","Ok",
NULL,NULL,fname,sys_errlist[errno]);
return(self);
}
/*--------------------------------< DELEGATE METHODS >-------------------------------*/
@end
These are the contents of the former NiCE NeXT User Group NeXTSTEP/OpenStep software archive, currently hosted by Netfuture.ch.