⟦0b4c8ea76⟧

TextFile

/* qmgr.c: Queue manager main routines */

# ifndef lint
static char Rcsid[] = "@(#)$Header: /cs/research/pp/hubris/pp-beta/Src/qmgr/RCS/qmgr.c,v 5.0 90/09/20 16:21:28 pp Exp Locker: pp $";
# endif

/*
 * $Header: /cs/research/pp/hubris/pp-beta/Src/qmgr/RCS/qmgr.c,v 5.0 90/09/20 16:21:28 pp Exp Locker: pp $
 *
 * $Log:	qmgr.c,v $
 * Revision 5.0  90/09/20  16:21:28  pp
 * rcsforce : 5.0 public release
 * 
 */



#include "util.h"
#include <sys/signal.h>
#include "ryresponder.h"                /* for generic idempotent responders */
#include "qmgr.h"                   /* operation definitions */
#include "types.h"

static char *myservice = "pp qmgr";
extern char *quedfldir;
static SFD debug_on (), debug_off (), shutdown ();
int ureject ();

/* OPERATIONS */
static int
	op_newmessage (), op_readmsginfo (),
	op_channelbegin (), op_channelread (),
	op_channelcontrol (), op_mtaread (), op_mtacontrol (),
	op_messagecontrol (), op_qmgrcontrol (), op_readchannelmtamessage ();


static struct server_dispatch dispatches[] = {
	"newmessage",       operation_Qmgr_newmessage,      op_newmessage,
	"readmsginfo",      operation_Qmgr_readmsginfo,     op_readmsginfo,
	"channelbegin",     operation_Qmgr_channelbegin,    op_channelbegin,
	"channelread",      operation_Qmgr_channelread,     op_channelread,
	"channelcontrol",   operation_Qmgr_channelcontrol,  op_channelcontrol,
	"mtaread",          operation_Qmgr_mtaread,         op_mtaread,
	"mtacontrol",       operation_Qmgr_mtacontrol,      op_mtacontrol,
	"msgcontrol",	    operation_Qmgr_msgcontrol,	    op_messagecontrol,
	"qmgrcontrol",	    operation_Qmgr_qmgrControl,	    op_qmgrcontrol,
	"readchannelmtamessage",
			    operation_Qmgr_readChannelMtaMessage,
						op_readchannelmtamessage,
	NULL
	};

extern int	chan_lose ();
extern int	start_routine ();
extern char *qmgr_hostname;

time_t	current_time;
time_t	debris_time = TRASH_RETRY_INTERVAL, load_time = LOAD_RETRY_INTERVAL;
time_t	cache_time = CACHE_TIME, timeout_time = TIMEOUT_RETRY_INTERVAL;
int	chan_state = 1;
int	maxchansrunning = MAXCHANSRUNNING;
int	submission_disabled = 0;
int	opmode = 0;

/* \f

   MAIN */

main (argc, argv)
int     argc;
char  **argv;
{
	int	opt;
	extern char *optarg;

	sys_init (myname = argv[0]);
	(void) time (&current_time);
	while ((opt = getopt (argc, argv, "Dd:l:c:t:m:s")) != EOF) {
		switch (opt) {
		    case 'D':
			chan_state = 0;
			break;

		    case 's':
			submission_disabled = 1;
			break;

		    case 'l':
			if ((load_time = atoi(optarg)) < 0)
				load_time = LOAD_RETRY_INTERVAL;
			else	load_time *= 60 * 60;
			break;

		    case 'd':
			if ((debris_time = atoi(optarg)) < 0)
				debris_time = TRASH_RETRY_INTERVAL;
			else	debris_time *= 60 * 60;
			break;

		    case 'c':
			if ((cache_time = atoi (optarg)) < 0)
				cache_time = CACHE_TIME;
			break;

		    case 't':
			if ((timeout_time = atoi (optarg)) < 0)
				timeout_time = TIMEOUT_RETRY_INTERVAL;
			else	timeout_time *= 60 * 60;
			break;

		    case 'm':
			if ((maxchansrunning = atoi (optarg)) < 1)
				maxchansrunning = MAXCHANSRUNNING;
			break;

		    default:
			adios (NULLCP,
	       "Usage: %s [-Ds] [-d hours] [-l hours] [-c secs] [-m nchans]",
			       myname);
			break;
		}
	}

#if defined(SIGUSR1) && defined(SIGUSR2)
	(void) signal (SIGUSR1, debug_on);
	(void) signal (SIGUSR2, debug_off);
#endif
	(void) signal (SIGTERM, shutdown);

	if (chdir (quedfldir) == NOTOK)
		PP_LOG (LLOG_EXCEPTIONS, ("can't cd to %s", quedfldir));
		
	ryresponder (argc, argv,  qmgr_hostname, myservice, dispatches,
		     table_Qmgr_Operations, start_routine,  chan_lose);

	exit (0);                       /* NOTREACHED */
}

/* \f

   OPERATIONS */

static int  op_newmessage (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	register struct type_Qmgr_MsgStruct *arg =
		(struct type_Qmgr_MsgStruct *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return add_msg (sd, arg, rox, roi);
}

static int op_readmsginfo (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	register struct type_Qmgr_ReadMessageArgument *arg =
		(struct type_Qmgr_ReadMessageArgument *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return read_msg (sd, arg, rox, roi);
}

static int op_channelbegin (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_FilterList *arg = (struct type_Qmgr_FilterList *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return chan_begin (sd, arg, rox, roi);
}

/* ARGSUSED */
static int op_channelread (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_UNIV_UTCTime *arg = (struct type_UNIV_UTCTime *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return channel_list (sd, arg, rox, roi);
}

static int op_mtacontrol (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_MtaControl *arg =
		(struct type_Qmgr_MtaControl *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return mta_control (sd, arg, rox, roi);
}

static int op_channelcontrol (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_ChannelControl *arg =
		(struct type_Qmgr_ChannelControl *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return chan_control (sd, arg, rox, roi);
}

static int op_mtaread (sd, ryo, rox, in, roi)
int     sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_MtaRead *arg = (struct type_Qmgr_MtaRead *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d: %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return mta_read (sd, arg, rox, roi);
}

static int op_messagecontrol (sd, ryo, rox, in, roi)
int	sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_MsgControl *arg = (struct type_Qmgr_MsgControl *)in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return msg_control (sd, arg, rox, roi);

}

static int op_qmgrcontrol (sd, ryo, rox, in, roi)
int	sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_QMGROp *op = (struct type_Qmgr_QMGROp *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return qmgrcontrol (sd, op -> parm, rox, roi);
}

static int op_readchannelmtamessage (sd, ryo, rox, in, roi)
int	sd;
struct RyOperation *ryo;
struct RoSAPinvoke *rox;
caddr_t in;
struct RoSAPindication *roi;
{
	struct type_Qmgr_MsgRead *mr = (struct type_Qmgr_MsgRead *) in;

	if (rox -> rox_nolinked == 0) {
		advise (LLOG_NOTICE, NULLCP,
			"RO-INVOKE.INDICATION/%d %s, unknown linkage %d",
			sd, ryo -> ryo_name, rox -> rox_linkid);
		return ureject (sd, ROS_IP_LINKED, rox, roi);
	}
	advise (LLOG_NOTICE, NULLCP, "RO-INVOKE.INDICATION/%d: %s",
		sd, ryo -> ryo_name);

	return msgread (sd, mr, rox, roi);
}
	

/*    ERROR */

int  error (sd, err, param, rox, roi)
int     sd,
	err;
caddr_t param;
struct RoSAPinvoke *rox;
struct RoSAPindication *roi;
{
	if (RyDsError (sd, rox -> rox_id, err, param, ROS_NOPRIO, roi) == NOTOK)
		ros_adios (&roi -> roi_preject, "ERROR");

	return OK;
}

int ureject (sd, reason, rox, roi)
int     sd, reason;
struct RoSAPinvoke *rox;
struct RoSAPindication *roi;
{
	if ( RyDsUReject (sd, rox -> rox_id, reason, ROS_NOPRIO, roi) == NOTOK)
		ros_adios (&roi -> roi_preject, "U-REJECT");
	return OK;
}

/* debug control */

static int	debugging = -1;

static SFD debug_on ()
{
	if (debugging == -1)
		debugging = pp_log_norm -> ll_events;
	if ((pp_log_norm -> ll_events & LLOG_TRACE) == 0)
		pp_log_norm -> ll_events |= LLOG_TRACE;
	else
		pp_log_norm -> ll_events |= LLOG_DEBUG;
}

static SFD debug_off ()
{
	if (debugging != -1) {
		pp_log_norm -> ll_events = debugging;
		debugging = -1;
	}
}

static SFD shutdown ()
{
	opmode = OP_SHUTDOWN;
}
DataMuseum.dk

DKUUG/EUUG Conference tapes

⟦0b4c8ea76⟧ TextFile

Derivation

TextFile
