/* 
 * $Id: s_handle.c,v 1.2 91/05/03 03:06:17 qjb Exp Locker: qjb $
 * $Source: /afs/athena.mit.edu/astaff/project/qcs/src/RCS/s_handle.c,v $
 * $Author: qjb $
 *
 * This source file contains the routines that handle the
 * different kinds of requests from the server's select loop.
 */

#if !defined(lint) && !defined(SABER) || defined(RCS_HDRS)
static char *rcsid = "$Id: s_handle.c,v 1.2 91/05/03 03:06:17 qjb Exp Locker: qjb $";
#endif /* !lint && !SABER || RCS_HDRS */


#include <stdio.h>
#include <syslog.h>
#include <sys/types.h>
#include <sys/param.h>
#include <sys/file.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <netdb.h>
#include <errno.h>
#include <krb.h>
#include <des.h>
#include <qrpc.h>

#include "qcs.h"
#include "qcs_private.h"
#include "qcs_qrpc.h"

static char *aname;
static char *keyfile;

static char tmpbuf[BUFSIZ];

void qcsi_init_aname_keyfile(char *aname1, char *keyfile1) 
{
    aname = aname1;
    keyfile = keyfile1;
}


void qcsi_handle_periodic(void)
{
    int i;
    qcs_client client;
    int max_fd;	
    unsigned long now;

    now = time(0);
    max_fd = qcsi_max_fd();

    for (i = 0; i <= max_fd; i++) {
	if (qcsi_fd_type(i) == qf_client) {
	    client = (qcs_client) qcsi_fd_data(i);
	    
	    if (client->nreq == 0) {
		/* Flush client if idle too long */
		if (now - client->last_req > QCS_FLUSHTIME)
		    qcsi_flush_client(&client);
	    }
	}
    }

    /* Notify all clients about the status of their jobs */
    qcsi_notify_jobs();
}


void qcsi_handle_connection(int fd)
{
    qrpc_error_t status;
    qrpc_t qrpc = NULL;
    qcs_client client = NULL;
    qrpc_t listen_qrpc;

    /*
     * Accept the new connection.  
     */

    if (qcsi_assert_fd_type(fd, "qcsi_handle_connection", qf_listen))
	return;
	
    listen_qrpc = (qrpc_t) qcsi_fd_data(fd);

    if (status = qrpc_accept(listen_qrpc, &qrpc)) {
	if (qrpc) {
	    sprintf(tmpbuf, 
		    "Error occured while accepting connection from %s: ",
		    qcsi_resolve_host(qrpc->caddr.sin_addr));
	    strcat(tmpbuf, qrpc_error_string(qrpc, status));
	    if (status == QRPC_ERR_VWARN)
		qcsi_error(LOG_WARNING, tmpbuf, NULL);
	    else
		qcsi_error(LOG_ERR, tmpbuf, NULL);
	}
	else
	    qcsi_error(LOG_ERR, qrpc_error_string(listen_qrpc, status), NULL);
	if (status != QRPC_ERR_VWARN) {
	    if (qrpc)
		qrpc_destroy(&qrpc);
	    return;
	}
    }

    sprintf(tmpbuf, "Accepted connection from %s", 
	    qcsi_resolve_host(qrpc->caddr.sin_addr));
    qcsi_debug(tmpbuf);

    /*
     * Create client object
     */
    if ((client = qcsi_get_client_object()) == NULL) {
	qrpc_drop(qrpc, "Not enough memory to accept new clients");
	qrpc_destroy(&qrpc);
	return;
    }

    /*
     * Initialize client's qrpc object, request time, and notify time
     */
    client->qrpc = qrpc;
    client->last_req = time(0);
    client->last_notified = client->last_req;

    /*
     * Try to read and decode authenticator 
     */
    if (status = qrpc_auth_from_client(qrpc, aname, keyfile, 
				       &(client->auth_dat),
				       client->sess_key, client->sched)) {
	sprintf(tmpbuf, "Failure accepting authentication: %s\n",
		qrpc_error_string(qrpc, status));
	qrpc_drop(qrpc, tmpbuf);
	qrpc_destroy(&qrpc);
	free((char *)client);
	return;
    }

    /*
     * Register this client's incoming file descriptor for select
     */
    
    client->id.fd = qrpc->in;
    qcsi_register_fd(qrpc->in, qf_client, (qcs_pointer) client);

    /* 
     * Log this connection.
     */
    sprintf(tmpbuf, "New connection from %s", qcsi_unparse_client(client));
    qcsi_log(tmpbuf);
}


void qcsi_handle_request(int fd)
{
    qcs_error_t status = QCS_SUCCESS;
    qcs_client client;
    qrpc_opcode_t opcode;
    qrpc_length_t length;

    /*
     * Find the client object associated with this file descriptor
     */

    if (qcsi_assert_fd_type(fd, "qcsi_handle_request", qf_client))
	return;

    client = (qcs_client) qcsi_fd_data(fd);

    /* Update the time that we've last heard from this client */
    client->last_req = time(0);
    client->last_notified = client->last_req;

    /* Get the qrpc packet header */
    if (status = qrpc_recv_header(client->qrpc, &opcode, &length)) {
	sprintf(tmpbuf, "Error getting header from client %s: %s",
		qcsi_unparse_client(client), 
		qrpc_error_string(client->qrpc, status));
	qcsi_error(LOG_NOTICE, tmpbuf, NULL);
	qcsi_flush_client(&client);
	return;
    }

    switch (opcode) {
      case QCS_OP_CMD:
	qcsi_get_command(client, length);
	break;
      case QCS_OP_QUIT:
	qcsi_client_quit(client);
	break;
      default:
	sprintf(tmpbuf, "application protocol error: unexpected opcode %d",
		opcode);
	qrpc_drop(client->qrpc, tmpbuf);
	qcsi_flush_client(&client);
	return;
    }
}


void qcsi_handle_completion(int fd)
{
    int pid;
    union wait wait_status;
    qcs_cmd_class_id cc_index;
    qcs_cmd_class *cmd_class;
    qcs_request request;
    qcs_client client;
    char job_type[2];
    
    if (qcsi_assert_fd_type(fd, "qcsi_handle_completion", qf_completion))
	return;
    
    job_type[1] = '\0';

    /* Read job type from file descriptor */
    if (read(fd, job_type, 1) == -1) {
	sprintf(tmpbuf, "failure reading from completion_fd: %s",
		sys_errlist[errno]);
	qcsi_error(LOG_CRIT, tmpbuf, NULL);
	return;
    }
    
    if (strcmp(job_type, QCS_JT_FORK) == 0) {
	switch (pid = wait(&wait_status)) {
	  case 0:
	    qcsi_error(LOG_CRIT, "wait() returned 0", NULL);
	    return;
	  case -1:
	    sprintf(tmpbuf, "wait failed: %s", sys_errlist[errno]);
	    qcsi_error(LOG_CRIT, tmpbuf, NULL);
	    return;
	  default:
	    break;
	}
    }
    else if (strcmp(job_type, QCS_JT_NOFORK) == 0) 
	pid = QCS_NOFORK;
    else {
	sprintf(tmpbuf, "Invalid job type %s sent over completion fd",
		job_type);
	qcsi_error(LOG_CRIT, tmpbuf, NULL);
	return;
    }

    if ((cc_index = qcsi_cmd_class_pid(pid)) == QCS_NO_CMD_CLASS)
	return;
	
    if (qcsi_dequeue_request(cc_index, &request))
	return;

    if ((cmd_class = qcsi_get_cmd_class(cc_index)) == NULL)
	return;

    if (cmd_class->pid != QCS_NOFORK)
	cmd_class->pid = QCS_READY;
    
    if (client = qcsi_get_client_from_req(&request)) {
	client->nreq--;
	client->last_req = time(0);
	
	sprintf(tmpbuf, "Completed command (%s) from client %s",
		qcsi_unparse_cmd(request.cmd), qcsi_unparse_client(client));
	qcsi_log(tmpbuf);

	/* 
	 * If this is a protected client, it must have just issued
	 * the shutdown command.  In this case, we do not want to send
	 * it a completion code so that we can continue to send status
	 * messages without the client thinking that the command has
	 * really completed.
	 */
	if (! client->shutdown) {
	    sprintf(tmpbuf, "Your command (%s) has completed.", 
		    qcsi_unparse_cmd(request.cmd));
	    if (qcsi_client_notify(client, tmpbuf)) 
		qcsi_flush_client(&client);

	    if (client) {
                /* Send a completion notice to the client */
		sprintf(tmpbuf, "Notifying client %s of completion of job %d",
			qcsi_unparse_client(client), request.jobno);
		qcsi_debug(tmpbuf);
                if (qcsi_client_notify_completion(client, request.jobno))
                    qcsi_flush_client(&client);
	    }
	}
	
	/* 
	 * Flush this client if it is done or if it is not protected
	 * and the server is going down.
	 */
	if (client && (client->done || 
		       (qcsi_is_server_shutdown() && 
			(! client->shutdown) &&
			client->nreq == 0)))			
	    qcsi_flush_client(&client);
    }
    
    free((char *)request.cmd);

    qcsi_try_dispatch(cc_index);
}