/* (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: distrib.c */
/* Author: David F. Bacon */
#ifndef lint
static char sccsinfo[] = "@(#)distrib.c	1.10 3/13/90";
#endif

#define _BSD 43

#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>

#include "asyncrpc.h"
#include "ops.h"
#include "distrib.h"
#include "storage.h"

/* don't ack each network op */
#undef PING

extern datarep dr_callmessage;

#define is_bottom(obj) ((obj)->tsdr is &dr_bottom)

#define HOSTNAMESIZE 512

schedblock *rpcsched;

interpname localaddr;


u_long interpno = 0;
object Linkportobj, *Linkport = &Linkportobj;

#ifdef DEBUG
#include <stdio.h>

#define DISTBUGLEVEL 7
char *hmsg_names[] = {
    "noop", 
    "call", "return", "return_error", "return_discarded", "return_exception",
    "send", "send_ok", "send_error",
    "adjust"
  };
#endif


void
init_distrib()
{
    extern object *Linkport;
    char hostname[HOSTNAMESIZE];

    struct hostent *hp;

    set_bottom(Linkport);

    if (gethostname(hostname, HOSTNAMESIZE) is -1)
      goto cleanup;
    if ((hp = gethostbyname(hostname)) is nil)
      goto cleanup;

    bcopy(hp->h_addr, (caddr_t) localaddr.hostname, HOSTIDSIZE);
    localaddr.number = 0;	/* filled in by start_network */
    return;

  cleanup:
    abort_nili("init_distrib");
}


status
compare_interpnames(i1, i2)
interpname *i1, *i2;
{
    if (strncmp(i1->hostname, i2->hostname, HOSTIDSIZE) is 0
	and i1->number is i2->number)
      return(SUCCESS);
    else
      return(FAILURE);
}


status
compare_remote_addresses(r1, r2)
remote_address *r1, *r2;
{
    if (compare_interpnames(& r1->interp, & r2->interp) is SUCCESS
	and r1->addr is r2->addr)
      return(SUCCESS);
    else
      return(FAILURE);
}
	

void
make_local_remaddr(remptr, addr)
remote_address *remptr;
remaddr addr;
{
    extern u_long interpno;

    remptr->addr = addr;
    bcopy(localaddr.hostname, (caddr_t) remptr->interp.hostname, HOSTIDSIZE);
    remptr->interp.number = interpno;
}


#ifdef DEBUG
void
debug_showop(sendrecv, opcode, proc)
char *sendrecv;
hmsg_command opcode;
pcb *proc;
{
    void showstatus();


    if (debug_level(DISTBUGLEVEL)) {
	fprintf(stderr, "%4s %12s: ", sendrecv, hmsg_names[(int) opcode]);
	showstatus(proc);
    }
}
#endif

/*****************************************************************************/
/*                      sending side routines                                */
/*****************************************************************************/

void sendrpc();

void
semiping(addr)
remote_address *addr;
{
    hmbuf_union u;

    sendrpc(addr, hmsg_noop, u);
}


/*ARGSUSED*/
predef_exception
remport_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;
{
    void re_finalize();
    hmbuf_union u;


    if (obj->tsdr->number is dr_callmessage.number) {
	/* bogus assumption: we could be here forwarding a callmessage */
	/* at best, this is just a storage leak */
	u.callmsg = obj->value;
	sendrpc(& port.outport->info.remote, hmsg_call, u);

	/* remove following when finalization stuff is done right */
	set_bottom(obj);

	return(Normal);		/* error comes back asynchronously */
    }
    else {			/* it's a send */
	u.sendmsg = *obj;
	sendrpc(& port.outport->info.remote, hmsg_send, u);
	re_finalize(obj, F_FREE, (schedblock *) nil);
	return(Normal);		/* fix later: what about Disconnected? */
				/*  if we implement it, we should block the */
				/*  process and wait for hmsg_send_ok */
    }
}


void
adjust_remport(dst, adjustment)
dfd_outport dst;
int adjustment;
{
    xdr_status hxdr_integer();
    hmbuf_union u;
    

    if (dst->info.remote.addr is LINKPORTADDR)
      return;

    u.adjustment = adjustment;
    sendrpc(& dst->info.remote, hmsg_adjust, u);
}


status
eq_remport(op1, op2)
valcell op1, op2;
{
    if (op1.outport is op2.outport)
      return(SUCCESS);		/* equal if pointing to same channel object */

    return(compare_remote_addresses(& op1.outport->info.remote,
				    & op2.outport->info.remote));
}


void
rem_fin_callmessage(value)
valcell value;
{
    hmbuf_union u;

    u.callmsg = value;

    sendrpc(& value.callmessage->info.callmessage->cminfo.remote, 
	    hmsg_return_discarded, u);
}

void
rem_return(cm)
valcell cm;
{
    hmbuf_union u;

    u.callmsg = cm;

    sendrpc(& cm.callmessage->info.callmessage->cminfo.remote, 
	    hmsg_return, u);
}


void 
rem_return_exception(cm, excep)
valcell cm, excep;
{
    hmbuf_union u;

    u.excepmsg.callmessage = cm;
    u.excepmsg.excep = excep;

    sendrpc(& cm.callmessage->info.callmessage->cminfo.remote, 
	    hmsg_return_exception, u);
}


void 
rem_return_error(cm, err)
valcell cm;
predef_exception err;
{
    hmbuf_union u;

    u.errormsg.callmessage = cm;
    u.errormsg.excep = err;

    sendrpc(& cm.callmessage->info.callmessage->cminfo.remote, 
	    hmsg_return_error, u);
}


/* not currently implemented due to use of rpc instead of sockets: */
/* return_error, send_ok, send_error */


void
sendrpc(target, opcode, data)
remote_address *target;
hmsg_command opcode;
hmbuf_union data;
{
    xdr_status hxdr_hmsg_buf();

    struct sockaddr_in server;
    int sock = RPC_ANYSOCK;
    CLIENT *cl;
    enum clnt_stat clnt_stat;
    hmsg_buf mbuf;


#ifdef DEBUG
    {
	extern schedblock schedb;

	debug_showop("send", opcode, schedb.ready);
    }
#endif

    mbuf.addr = target->addr;
    mbuf.sender = localaddr;
    mbuf.command = opcode;
    mbuf.mbuf = data;

    bcopy(target->interp.hostname, (caddr_t) & server.sin_addr, HOSTIDSIZE);
    server.sin_family = AF_INET;
    server.sin_port = 0;

    if ((cl = clntatcp_cachedcreate(&server, HERMESPROG, target->interp.number,
			       &sock, 0, 0)) is NULL) {
	clnt_pcreateerror("sendrpc");
	abort_nili("sendrpc");
    }

    clnt_stat = clnt_send(cl, HERMESPROC, hxdr_hmsg_buf, &mbuf);

    if (clnt_stat isnt RPC_SUCCESS) {
	clnt_perror(cl, "sendrpc");
	abort_nili("sendrpc");
    }
}


/*****************************************************************************/
/*                              xdr stuff                                    */
/*****************************************************************************/

xdr_status
hxdr_hmsg_buf(xdrs, mb)
XDR *xdrs;
hmsg_buf *mb;
{
    if (not xdr_opaque(xdrs, (char *) & mb->addr, sizeof(remaddr)))
      return(XDR_FAIL);
    if (not hxdr_interpname(xdrs, & mb->sender))
      return(XDR_FAIL);
    if (not xdr_enum(xdrs, (enum_t *) & mb->command))
      return(XDR_FAIL);

    switch (mb->command) {
      case hmsg_call:
      case hmsg_return:
      case hmsg_return_discarded: 
	return(hxdr_callmessage(xdrs, & mb->mbuf.callmsg));

      case hmsg_send: 
	return(hxdr_object(xdrs, & mb->mbuf.sendmsg));

      case hmsg_adjust:
	return(xdr_int(xdrs, & mb->mbuf.adjustment));

      case hmsg_return_exception: {
	  if (not hxdr_callmessage(xdrs, & mb->mbuf.excepmsg.callmessage))
	    return(XDR_FAIL);
	  return(hxdr_record(xdrs, & mb->mbuf.excepmsg.excep));
      }

      case hmsg_return_error: {
	  if (not hxdr_callmessage(xdrs, & mb->mbuf.errormsg.callmessage))
	    return(XDR_FAIL);
	  return(xdr_enum(xdrs, & mb->mbuf.errormsg.excep));
      }

      case hmsg_noop: 
	return(XDR_OK);

      default: {		/* hmsg_send_ok or hmsg_send_error */
	  abort_nili("hxdr_hmsg_buf");
	  /*NOTREACHED*/
      }
    }
}

/*****************************************************************************/
/*                      callee (server) side routines                        */
/*****************************************************************************/

#ifdef DEBUG
static pcb *newproc;
#endif

void
recvrpc(msg)
hmsg_buf *msg;
{
    valcell neta2inport(), neta2callmsg();
    void hrpc_call(), hrpc_return(), hrpc_return_discarded(),
      hrpc_return_exception(), hrpc_send(), hrpc_adjust(), hrpc_return_error();

#ifdef DEBUG
    newproc = nil;
#endif

    switch (msg->command) {
      case hmsg_call: {
	  hrpc_call(neta2inport(msg->addr), msg->mbuf.callmsg);
	  break;
      }
      case hmsg_return: {
	  hrpc_return(neta2callmsg(msg->addr));
	  break;
      }
      case hmsg_return_discarded: {
	  hrpc_return_discarded(neta2callmsg(msg->addr));
	  break;
      }
      case hmsg_return_exception: {
	  hrpc_return_exception(neta2callmsg(msg->addr), 
				msg->mbuf.excepmsg.excep);
	  break;
      }
      case hmsg_return_error: {
	  hrpc_return_error(neta2callmsg(msg->addr), msg->mbuf.errormsg.excep);
	  break;
      }
      case hmsg_adjust: {
	  hrpc_adjust(neta2inport(msg->addr), msg->mbuf.adjustment);
	  break;
      }
      case hmsg_send: {
	  hrpc_send(neta2inport(msg->addr), & msg->mbuf.sendmsg);
	  break;
      }
      case hmsg_noop: 
	break;

      default: 
	abort_nili("recvrpc");
    }

#ifdef DEBUG
    debug_showop("recv", msg->command, newproc);
#endif

#ifdef PING
    if (msg->command isnt hmsg_noop) {
	remote_address addr;

	addr.interp = msg->sender;
	addr.addr = 0;
	semiping(&addr);
    }
#endif
}


valcell
neta2callmsg(addr)
remaddr addr;
{
    valcell v;

    v.callmessage = (dfd_callmessage *) addr;
    return(v);
}

valcell
neta2inport(addr)
remaddr addr;
{
    extern object *Linkport;
    valcell v;

    if (addr is 0) {
	if (is_bottom(Linkport))
	  v.inport = nil;
	else
	  v = Linkport->value;
    }
    else
      v.inport = (dfd_inport *) addr;

    return(v);
}


void
hrpc_call(ip, callmsg)
valcell ip;
valcell callmsg;
{
    message *newmsg;
    object obj;
    predef_exception rc;

    if (ip.inport is nil or ip.inport->disconnected) {
	rem_return_error(callmsg, Disconnected);
	return;
    }

    if ((newmsg = new(message)) is nil) {
	rem_fin_callmessage(callmsg);
	return;
    }

    obj.value = callmsg;
    obj.tsdr = & dr_callmessage;

    rc = (*ip.inport->port_enq)(ip, & obj, newmsg, rpcsched);

    if (rc isnt Normal) 
      rem_return_error(callmsg, rc);
}


void
hrpc_return(cm)
valcell cm;
{
    pcb *do_return();

    pcb *caller;


    caller = do_return(cm);
    if (not caller->interpreter)
      caller->ip++;
#ifdef DEBUG
    newproc = caller;
#endif
    rpcsched->wakeup_proc(rpcsched, caller);
}


void
hrpc_return_discarded(cm)
valcell cm;
{
    pcb *fin_callmessage();


    fin_callmessage(cm, F_DISCARD, rpcsched);
#ifdef DEBUG
    newproc = cm.callmessage->info.callmessage->cminfo.local.caller;
#endif
}


void
hrpc_return_error(cm, excep)
valcell cm;
predef_exception excep;
{
    void raise_exception();
    pcb *do_return();

    pcb *caller;


    caller = do_return(cm);
    raise_remote_predefined(excep, caller);
#ifdef DEBUG
    newproc = caller;
#endif
    rpcsched->wakeup_proc(rpcsched, caller);
}


void
hrpc_return_exception(cm, excep)
valcell cm;
valcell excep;
{
    void raise_user();
    pcb *do_return();

    pcb *caller;


    caller = do_return(cm);
    raise_user(excep, caller);
#ifdef DEBUG
    newproc = caller;
#endif
    rpcsched->wakeup_proc(rpcsched, caller);
}


/* fix later: deal with errors */
void
hrpc_send(ip, obj)
valcell ip;
object *obj;
{
    message *newmsg;


    if (ip.inport is nil or ip.inport->disconnected) {
	return;
	/* return(Disconnected); */
    }

    if ((newmsg = new(message)) is nil) {
	return;
	/* return(Depletion); */
    }

    (void) (*ip.inport->port_enq)(ip, obj, newmsg, rpcsched);
				/* for now, ignore the error */
}


void
hrpc_adjust(ip, adjustment)
valcell ip;
int adjustment;
{
    pcb *cleanup_channel();

    if (ip.inport is nil) {
	nilerror("hrpc_adjust", 
		 "trying to adjust refcount of uninitialized linkport");
	return;
    }

    ip.inport->refcount = ip.inport->refcount + adjustment;
    if (ip.inport->refcount is 0)
      cleanup_channel(ip.inport, rpcsched);
}
