message.c

T-kernel 的extension源代码

/*
 *----------------------------------------------------------------------
 *    T-Kernel / Standard Extension
 *
 *    Copyright (C) 2006 by Ken Sakamura. All rights reserved.
 *    T-Kernel / Standard Extension is distributed 
 *      under the T-License for T-Kernel / Standard Extension.
 *----------------------------------------------------------------------
 *
 *    Version:   1.00.00
 *    Released by T-Engine Forum(http://www.t-engine.org) at 2006/8/11.
 *
 *----------------------------------------------------------------------
 */

/*
 *	msgmgr.c (msg)
 *
 *	Message management
 */

#include <basic.h>
#include <tk/tkernel.h>
#include <tk/util.h>
#include <extension/message.h>
#include <extension/errno.h>
#include <sys/pinfo.h>
#include <sys/util.h>
#include <sys/syslog.h>
#include <longlong.h>
#include <libstr.h>
#include <extension/sys/tkse_ssid.h>
#include <extension/sys/svc/ifmessage.h>
#include <sys/tkseconf.h>

#define	ME(q)			((MSGENT*)(q))
#define	TME(q)			((TOUTMSG*)(q))
#define	TEVMSK(tskevt)		( 1 << ((tskevt) - 1) )
#define	MM_NOPRC		0x80000000U	/* Indicates that process is in the course of termination processing. */

#define TSD_TBM_WAI_M1		(-1)
#define TSD_TDM_VAL_31		31
#define TSD_TDM_VAL_2		2
#define TSD_TRM_VAL_2		2
#define TSD_ITM_ITS_137		137
#define TSD_ITM_STK_512		512
#define TSD_MSE_CAS_1		1
#define TSD_MSE_CAS_2		2
#define TSD_MSE_CAS_3		3
#define TSD_MSE_CAS_4		4
#define TSD_MSE_CAS_5		5
#define TSD_MSE_CAS_6		6
#define TSD_MSE_CAS_7		7
#define TSD_MSE_CAS_8		8
#define TSD_MSE_CAS_9		9
#define TSD_MSG_MMS_8192	8192
#define TSD_MSG_TMM_65536	65536

IMPORT ER _tkse_abo_prc(ID pid, W abort_code);	/* prcmgr */
IMPORT QUEUE UsedPI;				/* PINFO queue in use 	*/
IMPORT W MAX_SUBTASKS;				/* Number of subtasks			*/

EXPORT W MAX_MSGSZ = TSD_MSG_MMS_8192;		/* Maximum size of individual message	*/
LOCAL W TOTAL_MSGMAX = TSD_MSG_TMM_65536;	/* Maximum size of message area	*/
LOCAL ID MsgAllocFlg;				/* Wait flag for obtaining message area	*/

/* Wait task specified by WAIEVT
 *	> 0	Wait task ID
 *	< 0	Reserved for reset of wait state
 *	= 0	No wait task nor reservation for reset of wait state
 *	As for access of waievt_tid, disable interrupt to perform exclusive control.
 *	By WAIEVT, it is possible to specify only one task at a time in the entire system.
 */
LOCAL ID waievt_tid = 0;

/* Message entry */
typedef	struct {
	QUEUE	q;			/* Queue			*/
	W	pid;			/* Send process PID		*/
	W	cfmtsk;			/* CNFM wait task ID	*/
	MESSAGE	msg;			/* Message body		*/
} MSGENT;

/* Timeout message entry  = Same as MESGENT in structure */
typedef	struct {
	QUEUE	q;			/* Queue			*/
	W	pid;			/* Intended process ID		*/
	W	cfmtsk;			/* CNFM wait task ID = 0	*/
	union	{			/* Timeout message body	*/
		struct {
			W	msg_type;
			W	msg_size;
		} m;
		longlong	time;	/* Absolute send time			*/
	} e;
	W	code;			/* Timeout code		*/
} TOUTMSG;

LOCAL QUEUE ToutMsgQue;		/* Timeout MSG queue		*/
LOCAL FastLock ToutMsgLock;	/* Timeout MSG lock		*/
LOCAL ID ToutMsgTskId;		/* Timeout MSG task ID		*/
LOCAL W MSG_TEV;		/* Message event		*/
LOCAL W BRK_TEV;		/* Message wait reset event	*/

LOCAL MSGENT *MsgAlloc(W msgsz, BOOL nowait);
LOCAL void MsgFree(MSGENT *me);
LOCAL ER MsgER_ERR(ER er);
LOCAL WER _tkse_rcv_msg(W t_mask, MESSAGE *msg, W msgsz, W opt);
LOCAL ER _tkse_clr_msg(W t_mask, W last_mask);
LOCAL ER _tkse_brk_msg( void );
LOCAL ER _tkse_DefMsgHdr(UW t_mask, FUNCP msg_hdr, FUNCP *hdrs);
LOCAL WER _tkse_RetMsgHdr(W abort);
LOCAL ER _tkse_req_tmg(TMOUT tmo, W code);
LOCAL void CancelToutMsg(W pid);
LOCAL ER _tkse_can_tmg(void);
LOCAL void TimeOutMsgTask( void );
LOCAL void InitToutMsg( void );
LOCAL WER MsgSVCentry(VP para, W fn);
LOCAL void MsgCleanUp(ID resid, W pid);
LOCAL void MsgStartUp(ID resid, W pid);
LOCAL void MsgBreak(ID taskid);

/*======================================================================
	Common processing
======================================================================*/
/*
	Obtain/release message area
		Maximum size of message area is TOTAL_MSGMAX.
*/
LOCAL	MSGENT	*MsgAlloc(W msgsz, BOOL nowait)
{
	MSGENT	*me;
	UW	dmy;

	msgsz = (W)(MSGSIZE((UW)msgsz) + offsetof(MSGENT, msg));
	for (;;) {
		me = (MSGENT*)Kmalloc((size_t)msgsz);
		if ((msgsz <= TOTAL_MSGMAX )&& me) {
			TOTAL_MSGMAX -= msgsz;
			return me;
		}
		if (nowait != 0) {
			return (MSGENT*)NULL;
		}
		tk_clr_flg(MsgAllocFlg, 0);
		tk_wai_flg(MsgAllocFlg, 1, TWF_ORW, &dmy, TMO_FEVR);
	}
}

/*
	Release message area
*/
LOCAL	void	MsgFree(MSGENT *me)
{
	TOTAL_MSGMAX += (W)(MSGSIZE((UW)me->msg.msg_size) + offsetof(MSGENT, msg));
	Kfree((VP)me);
	tk_set_flg(MsgAllocFlg, 1);
}

/*
	Message/error code conversion
*/
LOCAL	ER	MsgER_ERR(ER er)
{
	if ( er >= 0 ) {
		return er;
	}

	switch(er) {
	  case E_RLWAI:
		return E_NOEXS;
	  case E_QOVR:
		return E_LIMIT;
	  case E_TMOUT:
	  case E_DISWAI:
	  case E_DLT:
	  case E_ID:
	  case E_NOEXS:
		return er;
	  default:
		/* nothing to do */
		break;
	}
	return	E_SYS;
}

/*======================================================================
	Message SVC processing
======================================================================*/
/*
	Message sending
		msgflg == 0 : Ordinary message
		       <  0 : Startup/termination message (internal)
		       >  0 (== MS_TMOUT): Timeout message (internal)
*/
EXPORT	ER	_tkse_snd_msg(ID pid, MESSAGE *msg, W opt, W msgflg)
{
	PINFO	*pi;
	MSGENT	*me;
	W	i, tymask, msgty;
	W	msz;
	ER	er;

	msgty = msgflg;
	if (msgty > 0) {
		me = (MSGENT*)msg;	/* Timeout message */
	} else {
		if (msgflg == 0) {
			/* Parameter check */
			if (((UW)opt & ~(UW)(NOWAIT | CONFM)) != 0) {
				return E_PAR;
			}
			if ((CheckSpaceR((VP)msg, (W)MSGSIZE(0))) ||
			    (CheckSpaceR((VP)&msg->msg_body, msg->msg_size))) {
				return E_MACV;
			}
			if ((msg->msg_type < MS_MIN )||( msg->msg_type > MS_MAX)) {
				return E_PAR;
			}
		}
		msgty = msg->msg_type;
		msz   = msg->msg_size;
		if ((msz < 0 )||( (W)MSGSIZE((UW)msz) > MAX_MSGSZ)) {
			return E_LIMIT;
		}

		/* Obtain message area */
		me = MsgAlloc(msz, (BOOL)((UW)opt & (UW)NOWAIT));
		if (!(me )) {
			return E_SYSMEM;
		}
		/* Copy message */
		memcpy((VP)&me->msg, (VP)msg, (size_t)MSGSIZE((UW)msz));
	}
	er = PidToPinfo(pid, &pi);
	if (er != 0) {
		goto EXIT1;
	}

	if (msgflg <= 0) {
		PINFO	*mypi;
		if (GetMyPinfo(&mypi) != 0) {
			/* Send from other source than process */
			if (((UW)opt & (UW)CONFM) != 0) {
				er = E_PAR;
				goto EXIT2;
			}
			me->pid = 1;	/* Set source as INIT */
			me->cfmtsk = 0;	/* No CONFM wait */
		} else {
			/* Send from process */
			if ((mypi == pi )&&( ((UW)opt & (UW)CONFM))) {
				er = E_ILUSE;
				goto EXIT2;
			}
			me->pid = mypi->procid;
			me->cfmtsk = (((UW)opt & (UW)CONFM) != 0) ? GetMyTid() : 0;
		}
	}

	/* If it matches with ignmask at the destination, terminate the process normally. */
	tymask = MSGMASK(msgty) & MM_ALL;
	if ((pi->msg_ignmask & (UW)tymask) != 0U ) {
		if ((pi->msg_ignmask & MM_NOPRC) != 0) {
			/* Destination process is in the course of termination processing. */
			er = E_NOEXS;
		} else {
			opt &= ~(UW)CONFM;		/* Unnecessary to specify CNFM */
			MsgFree(me);
		}
		goto EXIT2;
	}

	/* Attach to the end of message queue at the destination */
	QueInsert(&me->q, &pi->msg_top);

	/* If it matches with mhmask at the destination, generate a forced exception. */
	if (((UW)tymask & pi->msg_mhmask) != 0U) {
		tk_ras_tex(pi->tsk[0].tskid, msgty);
		opt &= ~(UW)CONFM;		/* Ignore CNFM specification */
	} else {
		/* Check wait masks at the destination in order, and when there is a matching message type,
			reset wait state of the corresponding task. */
		for (i = 0; i < MAX_SUBTASKS; i++) {
			if ((pi->tsk[i].msg_waimask & (UW)tymask) != 0U) {
				pi->tsk[i].msg_waimask = 0;
				tk_sig_tev(pi->tsk[i].tskid, MSG_TEV);
				break;
			}
		}
	}
EXIT2:
	UnlockPinfo();
EXIT1:
	if (er < 0) {
		if (msgflg <= 0) {
			MsgFree(me);
		}
	} else if ((opt & CONFM) != 0) {
		/* If CNFM is specified, put into wait state. When it is reset, clear cfmtsk within the message;
		   if the queue has been cleared, release the area. */
		er = MsgER_ERR( tk_wai_tev(TEVMSK(MSG_TEV), TMO_FEVR) );
		me->cfmtsk = 0;
		if (! me->q.prev) {
			MsgFree(me);
		}
	}
	return er;
}

/*
	Send message to message handlers
*/
EXPORT	WER	_snd_hdr_msg(MESSAGE *msg, W opt)
{
	W	tymask, msgty, msz;
	ER	er;
	MSGENT	*me;
	QUEUE	*que;

	/* Parameter check */
	if (((UW)opt & ~(UW)(NOWAIT | CONFM)) != 0U) {
		return E_PAR;
	}
	if (CheckSpaceR((VP)msg, (W)MSGSIZE(0)) ||
	    CheckSpaceR((VP)&msg->msg_body, msg->msg_size)) {
		return E_MACV;
	}

	msz = msg->msg_size;
	if ((msz < 0 )||( (W)MSGSIZE((UW)msz) > MAX_MSGSZ)) {
		return E_LIMIT;
	}

	msgty = msg->msg_type;
	if ((msgty < MS_MIN )||( msgty > MS_MAX)) {
		return E_PAR;
	}
	tymask = (W)MSGMASK(msgty);

	/* Send message to all processes for which
			corresponding message handlers are defined. */
	LockPinfo();
	er = E_NOEXS;

	for (que = UsedPI.next; que != &UsedPI; que = que->next) {
		if ((((PINFO*)que)->msg_ignmask & (UW)tymask) != 0) {
			/* If it matches with ignmask, do not send message,
					but suppose that it has been sent. */
			if (er == E_NOEXS) {
				er = 0;
			}

		} else if ((((PINFO*)que)->msg_mhmask & (UW)tymask) != 0U) {
			/* If it matches with mhmask, send message. */

			/* Obtain message area */
			me = MsgAlloc(msz, TRUE);
			if (!(me )) {
				er = E_SYSMEM;
				break;
			}

			/* Copy message */
			memcpy((VP)&me->msg, (VP)msg, (size_t)MSGSIZE((UW)msz));
			me->pid = 1;		/* Set source as INIT */
			me->cfmtsk = 0;		/* No CNFM wait	*/

			/* Attach to the end of message queue at the destination */
			QueInsert(&me->q, &((PINFO*)que)->msg_top);

			/* Generate a forced exception. */
			tk_ras_tex(((PINFO*)que)->tsk[0].tskid, msgty);
			er = 1;
		}
	}
	UnlockPinfo();
	return er;
}

/*
	Receive message
	* When set at "t_mask == 0", wait for brk_msg() only.
	* When STARTMSG is specified as opt, fetch the first message
	  regardless of t_mask.
*/
LOCAL	WER	_tkse_rcv_msg(W t_mask, MESSAGE *msg, W msgsz, W opt)
{
	PINFO	*mypi;
	QUEUE	*q, *top;
	ER	er;
	W	i, n, waimask = 0;
	UINT	tevmsk;
	ID	mytid = GetMyTid();

	/* Parameter check */
	if (((UW)opt & ~(UW)(NOWAIT|CHECK|NOCLR|WAIEVT|STARTMSG)) != 0U) {
		return E_PAR;
	}
	if (((opt & WAIEVT) != 0) ? (t_mask < 0): (t_mask <= 0)) {
		return E_PAR;
	}
	if (msgsz < (W)MSGSIZE(0)) {
		return E_PAR;
	}
	if (CheckSpaceRW((VP)msg, msgsz) != 0) {
		return E_MACV;
	}

	er = PidToPinfo(0, &mypi);
	if (er != 0) {
		return er;
	}
AGAIN:
	er = 0;
	/* Search for t_mask message. */
	for (top = &mypi->msg_top, q = top->next; q != top; q = q->next) {
		if ((((UW)opt & (UW)STARTMSG) == 0U )
		 && ((MSGMASK(ME(q)->msg.msg_type) & (UW)t_mask) == 0U)) {
			continue;
		}

		/* Intended message exists. */
		n = (W)MSGSIZE((UW)ME(q)->msg.msg_size);
		memcpy((VP)msg, (VP)&(ME(q)->msg), (size_t)((n < msgsz) ? n : msgsz));

		if ( msgsz < n ) {
			er = E_PAR;	/* The entire message cannot be received. */
			break;
		}

		if (!(opt & NOCLR)) {
			QueRemove(q);
			q->prev = (QUEUE *)NULL;
		}

		er = ME(q)->pid;	/* Source PID */

		/* If message is waiting for CNFM, reset wait state of the other party. */
		if (ME(q)->cfmtsk != 0) {
			tk_sig_tev(ME(q)->cfmtsk, MSG_TEV);
		} else if (!(opt & NOCLR)) {
			MsgFree(ME(q));
		}
		break;
	}

	if ((opt & WAIEVT) != 0) {
		UW	psw;

		DI(psw);
		if ( waievt_tid < 0 ) {
			/* Reserved for reset of wait state */
			waievt_tid = 0;
			tevmsk = 0;
		} else {
			/* Not reserved for reset of wait state */
			waievt_tid = mytid;
			tevmsk = (t_mask == 0)? (UINT)TEVMSK(BRK_TEV):
					(UINT)(TEVMSK(MSG_TEV) | TEVMSK(BRK_TEV));
		}
		EI(psw);
	} else {
		tevmsk = (UW)TEVMSK(MSG_TEV);
	}

	if (er == 0) {		/* Intended message does not exist. */

		if ((opt & NOWAIT) ||( tevmsk == 0U)) {
			er = E_TMOUT;
		}
		if ((opt & CHECK) != 0) {
			/* Fetch the first message: always NOCLR, NOCNFM */
			if (isQueEmpty(top) != 0) {
				waimask = MM_ALL;
			} else {
				q = top->next;
				n = (W)MSGSIZE((UW)ME(q)->msg.msg_size);
				memcpy((VP)msg, (VP)&(ME(q)->msg),
							(size_t)((n < msgsz) ? n : msgsz));
				er = ME(q)->pid;
			}
		} else {
			waimask = t_mask;
		}

		if (er == 0) {
			/* Set intended wait mask and put into wait state. */
			for (i = 0; i < MAX_SUBTASKS; i++) {
				if (mypi->tsk[i].tskid == mytid) {
					break;
				}
			}
			mypi->tsk[i].msg_waimask = (UW)waimask;
			UnlockPinfo();
			er = MsgER_ERR( tk_wai_tev((INT)tevmsk, TMO_FEVR) );
			LockPinfo();
			if (er >= E_OK) {
				if (((UW)er & tevmsk & (UW)TEVMSK(MSG_TEV)) != 0) {
					goto AGAIN;
				}
				er = E_TMOUT; /* Reset of wait state */
			}
		}
	}
	if (er == mypi->procid) {
		er = 0;	/* Message from current process */
	}
	UnlockPinfo();
	return er;
}

#if _USE_TKSE_SNR_MSG
/*
	Send/receive message
		Since there is no problem if the current process is specified, do not return error.
*/
LOCAL	WER	_tkse_snr_msg(ID pid, MESSAGE *s_msg, W t_mask, MESSAGE *r_msg,
								W msgsz)
{
	ER	er;

	if ((er = _tkse_snd_msg(pid, s_msg, WAIT, 0)) != 0) {
		return er;
	}
	return	_tkse_rcv_msg(t_mask, r_msg, msgsz, WAIT | CLR);
}
#endif

/*
	Clear message
*/
LOCAL	ER	_tkse_clr_msg(W t_mask, W last_mask)
{
	QUEUE	*q, *top, *prev;
	PINFO	*mypi;
	W	tymask;
	ER	er;
	BOOL	clear_one = FALSE;

	/* Parameter check */
	if ((t_mask <= 0 )||( last_mask < 0)) {
		return E_PAR;
	}
	if (last_mask == MM_ALL) {
		clear_one = TRUE;
		last_mask = MM_NULL;
	}