/* (C) Copyright International Business Machines Corporation 23 January */
/* 1990.  All Rights Reserved. */
/*  */
/* See the file USERAGREEMENT distributed with this software for full */
/* terms and conditions of use. */
/* File: o_ports.c */
/* Author: David F. Bacon */
#ifndef lint
static char sccsinfo[] = "@(#)o_ports.c	1.19 3/13/90";
#endif

/* 
 * o_ports.c
 *
 * This file contains the NIL ops for ports.  These are:
 *    new_inport
 *    connect
 *    send
 *    receive(*)
 *    empty(*)
 *    discard_outport(*)
 *    discard_inport(*)
 * (*) does not raise Depletion.
 *
 * and the generic operations:
 *    fin_inport
 *    fin_outport
 *    cp_outport
 *    eq_inport
 *    eq_outport
 *
 * Call and return are defined in o_call.c.
 */

#include "ops.h"
#include "recursiv.h"
#include "storage.h"
#include "procenv.h"
#include "sysdep.h"
#include "accessors.h"

#include "predefined.cd"
#include "interpform.cd"

#define Dst (DstObj->value)
#define Src (SrcObj->value)
#define Src1 (Src1Obj->value)
#define Src2 (Src2Obj->value)

extern datarep dr_inport, dr_outport, dr_callmessage;


NILOP(o_new_inport)
{
    void re_finalize();
    dfd_inport *make_inport();

    dfd_inport *ip;
    extern flag cherm_flag;


    if ((ip = make_inport()) is nil)
      raise(Depletion);
    else {
        if (not cherm_flag)
          re_finalize(DstObj, F_DISCARD, args->sched);
				/* finalize the value of the destination; */

	Dst.inport = ip;
	set_init(DstObj, dr_inport);
    }
}
	


dfd_inport *
make_inport()
{
    predef_exception inport_enq();
    void ip_enqueuer();

    register dfd_inport *ip;
    
    
				/* allocate storage for the inport */
    if ((ip = new(dfd_inport)) isnt nil) {

	/* initialize channel-generic info */

	ip->type = LocalInport;	/* just a plain ol' inport */
	ip->port_enq = inport_enq; /* function to enqueue messages */
	ip->disconnected = FALSE; /* FALSE means the inport is init. */
	ip->refcount = 0;	/* no connected outports yet. */

	/* initialize inport-specific info */
	
	ip->info.inport.enqueuer = ip_enqueuer;	
				/* set enqueueing function to default */

	ip->info.inport.queue = nil; /* make an empty queue */
	ip->info.inport.tail = nil;
	ip->info.inport.tsdr = nil; /* unknown queue element type */

	ip->info.inport.waiting_owner = nil;
				/* initially, no one is waiting on an event */
                                /*  at this inport. */
    }
    return(ip);
}


NILOP(o_connect)
{
    void re_finalize();
    extern flag cherm_flag;


    OPCHK(SrcObj,inport);
    if (Src.inport->refcount is MAXCOUNTER) {
				/* check that we can increment counter. */
	raise(Depletion);	/* nope -- error. */
    }
    else {
	Src.inport->refcount++;	/* increment reference count of channel. */

        if (not cherm_flag)
	  re_finalize(DstObj, F_DISCARD, args->sched);
				/* finalize the value of the destination; */

	Dst.outport = Src.inport; /* outport points to the channel. */
	set_init(DstObj, dr_outport);
				/* set typestate and datarep */
    }
}


NILOP(o_send)
{
    void re_finalize();

    register object *msg;
    register dfd_outport op;
    message *newmsg;
    predef_exception err;

#ifdef OPCHECK
    assert(OPISTYPE(DstObj,outport) or OPISTYPE(DstObj,procport));
#endif
    msg = SrcObj;

    op = Dst.outport;

    if (op->disconnected) {
	raise(Disconnected);
	return;
    }

    if (op->type is LocalInport) {
	if (msg->tsdr->number is dr_inport.number and
	    msg->value.inport is op) {
				/* detect black holes where an inport is */
                                /*  being sent to itself. */

	    re_finalize(SrcObj, F_DISCARD, args->sched);
				/* a black hole discards the inport */
	    return;
	}
    }


    if ((newmsg = new(message)) is nil)
      raise(Depletion);
    else if ((err = (*op->port_enq)(Dst, msg, newmsg, args->sched))
	     isnt Normal)
      raise(err);
    else {
#ifdef TRACE
	{
				/* fix call-level if fwding a cmsg */
	    if (msg->tsdr->number is dr_callmessage.number)
	      args->sched->ready->call_level = 0;
	}
#endif
    }
}



predef_exception
inport_enq(port, obj, msg, sched)
valcell port;			/* the outport to enq on. */
object *obj;			/* the object to send. */
message *msg;			/* pre-allocated message object. */
schedblock *sched;
{
    (*port.outport->info.inport.enqueuer)(port.outport, obj, msg, sched);
    return(Normal);
}


predef_exception
procedure_enq(port, obj, msg, sched)
valcell port;			/* the procport to enq on. */
object *obj;			/* the object to send. */
message *msg;			/* pre-allocated message object. */
schedblock *sched;
{
    pcb *create_proc();

    pcb *proc;
    dfd_inport *initport;

    proc = port.outport->info.procedure;

    if (proc->type is ProcedurePrimaryFree and
	not isbottom(&proc->ep.h->data[INITPORT]))
      proc->type = ProcedurePrimaryUsed;
    else			/* already in use... spawn another instance */
      if ((proc = create_proc(proc->prog)) is nil)
	return(Depletion);
      else
	proc->type = ProcedureSecondary;

    initport = dcdot(proc->ep.h, INITPORT).inport;

    /* could fold call-receive here for optimization purposes */

    /* send the callmessage */
    (*initport->info.inport.enqueuer)(initport, obj, msg, sched);
    
    sched->add(sched, proc);	/* start scheduling the new process */

    return(Normal);
}


void
ip_enqueuer(chan, obj, msg, sched)
dfd_inport *chan;
object *obj;
message *msg;
schedblock *sched;
{
    pcb *waiter;
    int i;
    dfd_inport *sleepip;
    local_inport *ip;

    ip = & chan->info.inport;

    waiter = nil;		/* no process re-awakened by this send. */
    
    ip->tsdr = obj->tsdr;       /* store typestate and data representation */
                                /*  in case we are the first message at this */
                                /*  inport. */

    msg->value = obj->value;    /* store the value in the message. */
    cdr(msg) = nil;
    
    if (ip->queue is nil) {	/* if this is the only message: */
        ip->queue = ip->tail = msg; /* make it the head and tail. */

	if (ip->waiting_owner) { /* if someone was waiting, revive him. */
	    waiter = ip->waiting_owner;
	    sched->wakeup_proc(sched, waiter);
	    ip->waiting_owner = nil;

	    if (waiter->selecting) { /* if sleeper was in a select... */
		waiter->selecting = FALSE;

		for (i = 0; i < waiter->suspend_info.select.opencount; i++) {
		    sleepip = waiter->suspend_info.select.waiting_ports[i];

		    if (sleepip is chan)
		      waiter->ip = waiter->suspend_info.select.branch_labels[i];

		    sleepip->info.inport.waiting_owner = nil;
		}
	    }			/* end code for selecting sleeper */
	}			/* end code for sleeper */
    }				/* end code for no messages on queue */

    else {                      /* else there are already messages enqueued; */
				/*  stick it at the end of the queue. */

        cdr(ip->tail) = msg;	/* append to tail... */
        ip->tail = msg;         /*  ....and make this one the tail. */
    }

    set_bottom(obj);		/* set typestate of message object to bottom */
				/*  in the sending process. */
}

/* does not raise Depletion */
NILOP(o_receive)
{
    void re_finalize();

    extern flag cherm_flag;

    register local_inport *ip;
    register object *obj;
    pcb *currentproc;
    message *nextmsg;
    

    OPCHK(SrcObj,inport);
    ip = & Src.inport->info.inport; /* get source inport pointer */
    
    if (ip->queue is nil) {     /* no queued messages; go to sleep. */

	if (Src.inport->refcount is 0) { 
				/* if no more messages will appear... */
	    raise(Disconnected); /* raise an exception */
	    return;
	}

        currentproc = args->sched->ready;
                                /* get a handle on the current process. */

        ip->waiting_owner = currentproc;
                                /* set inport to be waiting on current proc. */
        args->nextop = currentproc->ip;
                                /* set process to resume this receive when */
                                /*  it wakes up. */
	args->sched->suspend(args->sched, currentproc, nil);
                                /* deschedule this process in mainloop. */
    }
    else {                      /* there are queued messages... de-q one. */

        obj = DstObj;
                                /* get pointer to destination object */

	if (not cherm_flag)
	  re_finalize(obj, F_DISCARD, args->sched);
				/* finalize the value currently in the */
				/*  destination; */

        obj->value = ip->queue->value;
                                /* extract the value... */
        obj->tsdr = ip->tsdr;   /*  ...and the typestate/representation. */

	nextmsg = cdr(ip->queue);
				/* save pointer to the next message. */

	if (ip->tsdr->number isnt dr_callmessage.number)
	  { dispose(ip->queue, message); }
				/* free storage used by message unless its a */
				/*  callmessage (since for callmessages it's */
				/*  pre-allocated. */

        ip->queue = nextmsg;	/* advance the queue */

        if (ip->queue is nil)
          ip->tail = nil;       /* fix an empty queue (even though it isnt */
				/*  strictly necessary). */
#ifdef TRACE
	/* set this proc's call level one higher than caller's */
	if (ip->tsdr->number is dr_callmessage.number) {
	  pcb *caller;

	  if (obj->value.callmessage->info.callmessage->local) {
	      caller = obj->value.callmessage->info.callmessage->
		cminfo.local.caller;
	      args->sched->ready->call_level = caller->call_level + 1;
	  }
	  else
	    args->sched->ready->call_level = 1;
	}
#endif

    }
}


NILOP(o_empty)
{
    /* does not raise depletion */

    OPCHK(SrcObj,inport);
    if (Src.inport->info.inport.queue is nil) 
      Dst.boolean = nil_true;
    else 
      Dst.boolean = nil_false;

    set_init(DstObj, dr_boolean);
}



void
fin_outport(op, f_op, sched)
valcell op;
finalize_op f_op;
schedblock *sched;
{
    void cleanup_channel();
    void adjust_remport();


    if (f_op is F_DISCARD or op.outport->type is RemotePort) {
				/* if it's a true discard or if it's a free */
				/*  of a remote port, really try to free it: */
	if (--op.outport->refcount is 0) 
				/* no more outports connected to it... */
	  cleanup_channel(op.outport, sched);
    }
    if (f_op is F_DISCARD and op.outport->type is RemotePort)
      adjust_remport(op.outport, -1);

    /* if it's F_FREE on a local outport, we don't do anything. */
}


/* called by fin_outport() and hrpc_adjust() */
void
cleanup_channel(chan, sched)
channel *chan;
schedblock *sched;
{
    void revive_waiter();
    void freepcb();
    void re_finalize();

    switch (chan->type) {
      case LocalInport: {
	  if (chan->disconnected) {
	      /* inport discarded as well? */
	      dispose(chan, channel);
	      /* yes.  trash it. */
	  }
	  else		/* inport still intact. */
	    if (chan->info.inport.waiting_owner)
	      /* might wake up waiting process. */
	      revive_waiter(& chan->info.inport, sched);
	  break;
      }
	
      case LocalProcport: {
	  if (chan->info.procedure->type is ProcedurePrimaryFree) {
	      re_finalize(&chan->info.procedure->ep.h->data[INITPORT],
			  F_DISCARD, sched);
	      freepcb(chan->info.procedure);
	      dispose(chan->info.procedure, pcb);
	      dispose(chan, channel);
	  }
	  else {		/* it's ProcedurePrimaryUsed */
	      chan->info.procedure->type = ProcedurePrimaryDead;
	      /* mark it for garbage collection on */
	      /*  termination. */
	      dispose(chan, channel);
	  }
	  break;
      }
	
      case RemotePort: {
	  dispose(chan, channel);
	  break;
      }
    }
}


static void
revive_waiter(ip, sched)
local_inport *ip;
schedblock *sched;
{
    void raise_remote_predefined();
    flag deadselect();

    pcb *waiter;


    waiter = ip->waiting_owner;

    if (not waiter->selecting or deadselect(waiter)) {
	raise_remote_predefined(Disconnected, ip->waiting_owner);
	if (not waiter->interpreter)
	  sched->wakeup_proc(sched, waiter);
    }
}


static flag
deadselect(proc)
pcb *proc;
{
    select_suspend_info *info;

    int i;


    info = & proc->suspend_info.select;

    for (i = 0; i < info->opencount; i++) 
      if (info->waiting_ports[i]->refcount isnt 0)
	return(FALSE);

    return(TRUE);		/* all inports have no outport connected */
}




/*ARGSUSED*/
void
fin_inport(chan, f_op, sched)
valcell chan;
finalize_op f_op;		/* not used since inports are only local */
schedblock *sched;
{
    message *msgp;
    local_inport *ip;

    ip = & chan.inport->info.inport;

    for (msgp = ip->queue; msgp isnt nil; msgp = msgp->next) 
      (*ip->tsdr->finalize)(msgp->value, f_op, sched);

    if (chan.inport->refcount is 0) {
	dispose(chan.inport, channel); 
    }
    else 
      chan.inport->disconnected = TRUE;
}


status
eq_inport(ip1, ip2)
valcell ip1, ip2;
{
    if (ip1.inport is ip2.inport)
      return(SUCCESS);		/* an inport only equals itself */
    else
      return(FAILURE);
}


status
eq_outport(op1, op2)
valcell op1, op2;
{
    status eq_remport();


    if (op1.outport->type isnt op2.outport->type)
      return(FAILURE);		/* must be same type to be equal */

    if (op1.outport->type is RemotePort)
      return(eq_remport(op1, op2));

    /* LocalInport or LocalProcport */
    if (op1.outport is op2.outport)
      return(SUCCESS);
    else
      return(FAILURE);
}


predef_exception
cp_outport(dst, src)
valcell *dst, src;
{
    void adjust_remport();


    if (src.outport->refcount is MAXCOUNTER) 
				/* check that we can increment counter. */
      return(Depletion);

    src.outport->refcount++;	/* increment counter of connected outports. */
    dst->outport = src.outport;	/* and make a new reference */

    if (src.outport->type is RemotePort)
      adjust_remport(dst->outport, 1);
				/* add 1 to remote refcount */

    return(Normal);
}
