mserver.c

一套客户/服务器模式的备份系统代码,跨平台,支持linux,AIX, IRIX, FreeBSD, Digital Unix (OSF1), Solaris and HP-UX.

	connstat->newfilerequested = NO;
    }

    i = write_to_server(connstat->tapebuf, MPX_CHUNKSIZE,
					fromfd, tofd, tcpmux_status);
    ER__(connstat->outbuf[0] = (UChar) i, i);

    server_status = connstat->tape_status = OPENED_WRITING;

    connstat->tapebufptr = connstat->headerlen;		/* reset pointer */

    tocb -= n;
    inbufptr += n;
  } while(tocb > 0);

  return(0);
}

static Int32
readfromtape(
  ConnStatus	**csp,
  Int32		idx,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  Int32		cycles = 0;
  Int32		i, j, n;
  UChar		*cptr, res = 0;
  Flag		moved_start_byte;
  ConnStatus	*connstat, **allcs;

  connstat = *csp;
  allcs = csp - idx;

  while((n = connstat->commbufsiz - connstat->tapebufptr) > 0){
    cycles++;			/* first a check for interrupted, useful, */
    if(cycles > 100){			/* if we ain't getting sync'd */
	cycles = 0;
	if(check_interrupted(connstat))
	  return(interrupted);
    }

    while(tapebufptr < MPX_CHUNKSIZE){
	ER__(send_cmd(tofd, READFROMTAPE), j);	/* need more from tape */
	i = commbufsiz + 1;			/* one more cause of result */
	j = tcp_mux_long_read(tcpmux_status, fromfd, tapebuf + tapebufptr, i);
	tapebufptr += commbufsiz;		/* now points to result byte */
	res = tapebuf[tapebufptr];
	if(j != i || res)
	  return(res ? res : - errno);
    }

    if(!reading_in_sync){
	moved_start_byte = NO;

	for(cptr = tapebuf; cptr < tapebuf + tapebufptr; cptr++){
	  if((Uns16)(*cptr) == 0xff){
	    for(csp = allcs; *csp; csp++){
		j = (*csp)->tape_status;
		if((j == OPENED_FOR_READ || j == OPENED_READING)
			&& (cptr - tapebuf) + 1 + strlen((*csp)->clientid) + 1 < tapebufptr
			&& (!memcmp(cptr + 1, (*csp)->clientid, strlen((*csp)->clientid) + 1))){
		  reading_in_sync = YES;
		  break;
		}
	    }

	    i = cptr - tapebuf;
	    if(i > 0){
		if(tapebufptr > i){
		  memmove(tapebuf, cptr, tapebufptr - i);
		  moved_start_byte = YES;
		}
		tapebufptr -= i;
		cptr -= i;
	    }

	    if(reading_in_sync){
		break;
	    }
	  }
	}

	if(!reading_in_sync){
	  if(!moved_start_byte)
	    tapebufptr = 0;
	  continue;
	}
    }

    if((Uns16)(tapebuf[0]) != 0xff){
	UChar	cbuf[200];

	reading_in_sync = NO;
	logmsg(LOG_ERR,
		T_("Error: Demultiplexing stream falling out of sync."
			" Current bytes in buffer:\n"));
	strcpy(cbuf, "");
	for(j = 0; j < 32; j++)
	  sprintf(cbuf + strlen(cbuf), "%02x ", (int) tapebuf[j]);
	i = strlen(cbuf);
	for(j = 0; j < 32; j++){
	  cbuf[i++] = (isprint(tapebuf[j]) ? tapebuf[j] : (UChar) '.');
	  cbuf[i] = '\0';
	}
	logmsg(LOG_ERR, "%s\n", cbuf);

	continue;
    }

    if(tapebufptr < MPX_CHUNKSIZE)
	continue;

    if(!strcmp(tapebuf + 1, connstat->clientid)){
	j = MPX_CHUNKSIZE - connstat->headerlen;
	ER__(resize_tapebuf(connstat, connstat->tapebufptr + j), i);

	memcpy(connstat->tapebuf + connstat->tapebufptr,
			tapebuf + connstat->headerlen, j);
	connstat->tapebufptr += j;
    }

    if(tapebufptr > MPX_CHUNKSIZE)
	memmove(tapebuf, tapebuf + MPX_CHUNKSIZE, tapebufptr - MPX_CHUNKSIZE);
    tapebufptr -= MPX_CHUNKSIZE;
  }

  memcpy(connstat->outbuf, connstat->tapebuf, connstat->commbufsiz);
  connstat->outbuf[connstat->commbufsiz] = COMMAND_OK;

  if(connstat->tapebufptr > connstat->commbufsiz)
    memmove(connstat->tapebuf, connstat->tapebuf + connstat->commbufsiz,
		connstat->tapebufptr - connstat->commbufsiz);

  connstat->tapebufptr -= connstat->commbufsiz;

  return(0);
}

static Int32
getpos(int fromfd, int tofd, void * tcpmux_status, UChar cmd)
{
  UChar		buf[8];
  Int32		*cartptr, *fileptr;
  Int32		e;

  if(cmd == QUERYWRPOSITION ? wrtapepos_valid
		: (cmd == QUERYRDPOSITION ? rdtapepos_valid : tapepos_valid))
    return(0);

  ER__(send_cmd(tofd, cmd), e);

  ER__((read_forced(fromfd, buf, 8) != 8), e);

  if(buf[7])
    return((Int32) buf[7]);

  cartptr = (cmd == QUERYWRPOSITION ? &wrcartnum
		: (cmd == QUERYRDPOSITION ? &rdcartnum : &cartnum));
  fileptr = (cmd == QUERYWRPOSITION ? &wrfilenum
		: (cmd == QUERYRDPOSITION ? &rdfilenum : &filenum));

  xref_to_UnsN(cartptr, buf, 24);
  xref_to_Uns32(fileptr, buf + 3);

  if(cmd == QUERYWRPOSITION)
    wrtapepos_valid = YES;
  else if(cmd == QUERYRDPOSITION)
    rdtapepos_valid = YES;
  else
    tapepos_valid = YES;

  return(0);
}

static Int32
setcart(Int32 cart, int fromfd, int tofd, void * tcpmux_status)
{
  UChar		buf[4];
  Int32		e;

  tapepos_valid = NO;

  buf[0] = (cart > 0 ? SETCARTRIDGE : SETRAWCARTRIDGE);
  UnsN_to_xref(buf + 1, ABS(cart), 24);

  ER__((tcp_mux_long_write(tcpmux_status, tofd, buf, 4) != 4), e);

  ER__(result(fromfd, tcpmux_status), e);

  return(getpos(fromfd, tofd, tcpmux_status, QUERYPOSITION));
}

static Int32
setfile(Int32 file, int fromfd, int tofd, void * tcpmux_status)
{
  UChar		buf[5];
  Int32		e;

  tapepos_valid = NO;

  buf[0] = (file > 0 ? SETFILE : SETRAWFILE);
  Uns32_to_xref(buf + 1, ABS(file));

  ER__((tcp_mux_long_write(tcpmux_status, tofd, buf, 5) != 5), e);

  ER__(result(fromfd, tcpmux_status), e);

  return(getpos(fromfd, tofd, tcpmux_status, QUERYPOSITION));
}

static Int32
gettapeblocksize(Int32 * size, int fromfd, int tofd, void * tcpmux_status)
{
  UChar		buf[5];
  Int32		e;

  ER__(send_cmd(tofd, QUERYTAPEBLOCKSIZE), e);

  ER__(tcp_mux_long_read(tcpmux_status, fromfd, buf, 4) != 4, e);

  xref_to_Uns32(size, buf);

  return(result(fromfd, tcpmux_status));
}

static Int32
set_tapebuf_size(Int32 newsize)
{
  if(newsize > tapebuf_allocated){
    tapebuf = ZRENEWP(tapebuf, UChar, newsize);

    if(!tapebuf)
	return(FATAL_ERROR);
  }
  tapebuf_allocated = newsize;

  return(0);
}

static Int32
set_commbufsiz(
  Int32		newsize,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  UChar		buf[6];
  Int32		e;

  ER__(set_tapebuf_size((newsize + 2) * 2), e);

  buf[0] = SETCOMMBUFSIZ;
  Uns32_to_xref(buf + 1, newsize);

  ER__((write_forced(tofd, buf, 5) != 5), e);

  commbufsiz = newsize;

  return(result(fromfd, tcpmux_status));
}

static Int32
set_pos_if_req(
  ConnStatus	*connstat,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  Int32		i;

  if(connstat->req_cart /* && connstat->req_cart != cartnum */){
    i = setcart(connstat->req_cart, fromfd, tofd, tcpmux_status);
    connstat->req_cart = 0;
    connstat->outbuf[0] = (UChar) i;
    if(i)
	return(i);
  }

  if(connstat->req_filenum /* && connstat->req_filenum != filenum */){
    i = setfile(connstat->req_filenum, fromfd, tofd, tcpmux_status);
    connstat->req_filenum = 0;
    connstat->outbuf[0] = (UChar) i;
    if(i)
	return(i);
  }

  return(0);
}

static Int32
set_req_streamermode(
  ConnStatus	*connstat,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  Int32		e;

  ER__(simple_command(connstat->req_streamermode & BUFFERED_OPERATION ?
	SETBUFFEREDOP : SETSERIALOP, fromfd, tofd, tcpmux_status), e);

  ER__(simple_command(connstat->req_streamermode & CHANGE_CART_ON_EOT ?
	SETCHCARTONEOT : SETERRORONEOT, fromfd, tofd, tcpmux_status), e);

  return(0);
}

static Int32
pending_close_ops(
  ConnStatus 	*connstat,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  Int32		i, l, n;

  switch(connstat->tape_status){
    case OPENED_FOR_WRITE:
    case OPENED_WRITING:
	if(connstat->tapebufptr > connstat->headerlen){
	  memset(connstat->tapebuf + connstat->tapebufptr, 0,
		sizeof(UChar) * (MPX_CHUNKSIZE - connstat->tapebufptr));
	  i = write_to_server(connstat->tapebuf, MPX_CHUNKSIZE,
					fromfd, tofd, tcpmux_status);
	  ER__(connstat->outbuf[0] = (UChar) i, i);
	}
	l = MPX_CHUNKSIZE - connstat->headerlen;
	memset(connstat->tapebuf + connstat->headerlen, 0, l);
	for(n = MAX_PROT_CHUNKSIZE; n > 0; n -= l){
	  i = write_to_server(connstat->tapebuf, MPX_CHUNKSIZE,
					fromfd, tofd, tcpmux_status);
	  ER__(connstat->outbuf[0] = (UChar) i, i);
	}
  }

  return(0);
}

static Int32
closetape(
  Uns32		cmd, 
  int		fd,
  ConnStatus	**csp,
  Int32		idx,
  void		*user_data,
  int		fromfd,
  int		tofd,
  void		*tcpmux_status)
{
  Int32		i;
  ConnStatus	*connstat, **allconns;

  connstat = *csp;

  if(connstat->tape_status == NOT_OPEN || connstat->tape_status == CLOSED_NOTREW)
    return(0);

  allconns = csp - idx;

  i = pending_close_ops(connstat, fromfd, tofd, tcpmux_status);

  connstat->tape_status = (cmd == CLOSETAPE ? NOT_OPEN : CLOSED_NOTREW);
  raw_mode = NO;

  if(server_status & OPENED_RAW_ACCESS){
    server_status = connstat->tape_status = NOT_OPEN;

    return(PROC_DEFAULT_PROC);
  }

  switch(server_status){
	case OPENED_FOR_WRITE:
	case OPENED_WRITING:
		numwrclients--;

		tapepos_valid = wrtapepos_valid = rdtapepos_valid = NO;

		if(numwrclients == 0){
		  i = flush_to_server(fromfd, tofd, tcpmux_status);
		  connstat->outbuf[0] = (UChar) i;

		  server_status = connstat->tape_status = NOT_OPEN;

		  free_prev_wrclient_list();

		  return(PROC_DEFAULT_PROC);
		}

		if(!i)
		  i = getpos(fromfd, tofd, tcpmux_status, QUERYWRPOSITION);

		need_new_tapefile_for_wrclient(connstat->clientid,
						cartnum, filenum);

		if(i)
		  return(i);

		break;

	case OPENED_FOR_READ:
	case OPENED_READING:
		numrdclients--;

		tapepos_valid = wrtapepos_valid = rdtapepos_valid = NO;

		if(numrdclients == 0){
		  server_status = connstat->tape_status = NOT_OPEN;

		  return(PROC_DEFAULT_PROC);
		}

		break;
  }

  return(0);
}

Int32
start_slave_server()
{
  int		inpipe[2], outpipe[2];
  Int32		i, n;

  if(pipe(inpipe) || pipe(outpipe)){
    logmsg(LOG_ERR, T_("Error: Cannot create pipes to subprocess.\n"));
    do_exit(6);
  }

  pid = fork_forced();
  if(pid < 0){
    logmsg(LOG_ERR, T_("Error: Cannot start subprocess.\n"));
    do_exit(7);
  }

  if(!pid){
    close(inpipe[1]);
    close(outpipe[0]);

    dup2(inpipe[0], 0);
    dup2(outpipe[1], 1);

    execvp(sprog, sargv);

    logmsg(LOG_ERR, T_("Error: Cannot start program `%s'.\n"), sprog);

    exit(99);
  }

  close(inpipe[0]);
  close(outpipe[1]);
  fromfd = outpipe[0];
  tofd = inpipe[1];

  if(set_cryptkey_for_host(cryptfile, ""))
    do_exit(15);

  i = logon_to_server(fromfd, tofd, GREETING_MESSAGE, VERSION_MESSAGE_KEY,
				NULL,
#ifdef	HAVE_DES_ENCRYPTION
			AUTH_USE_DES
#else
			0
#endif
			);
  if(i){
    logmsg(LOG_ERR, T_("Error: Cannot logon to server ?!?!?\n"));
    do_exit(8);
  }

  n = (MPX_CHUNKSIZE + 2) * 2;
  ER__(set_tapebuf_size(n), i);
  memset(tapebuf, 0, n);

  i = getpos(fromfd, tofd, NULL, QUERYPOSITION);

  i = gettapeblocksize(&tapeblocksize, fromfd, tofd, NULL);
  i = set_commbufsiz(MIN(MAXCOMMBUFSIZ, tapeblocksize), fromfd, tofd, NULL);

  return(i);
}

/* The protocol handling routines */
static void *
init_conn_func(
  int		fd,
  Int32		idx,
  void		*allconns,
  struct sockaddr	*addr,
  void		*data,
  TcpMuxCallbDoneActions	*ret_actions)
{
  Int32		i;
  ConnStatus	*connstat;
  UChar		*peername = NULL;

  if(daemonize && pid < 0){
    i = start_slave_server();

    if(i)
	return(NULL);
  }

  peername = get_hostnamestr(addr);
  if(set_cryptkey_for_host(cryptfile, peername)){
    ZFREE(peername);
    return(NULL);
  }

  i = authenticate_client(fd, fd, unsecure,
			VERSION_MESSAGE GREETING_MESSAGE, 0, AUTHENTICATION,
#ifdef	HAVE_DES_ENCRYPTION
			AUTH_USE_DES
#else
			0
#endif
			, 0);
  if(i){
    ZFREE(peername);
    return(NULL);
  }

  connstat = NEWP(ConnStatus, 1);
  if(!connstat){
    ZFREE(peername);
    return(NULL);
  }

  SETZERO(connstat[0]);
  connstat->status = CONNSTAT_CONNECTED;
  connstat->tape_status = NOT_OPEN;
  connstat->fd = fd;
  connstat->inbuf = NEWP(UChar, MAX_PROT_CHUNKSIZE);
  connstat->outbuf = NEWP(UChar, MAX_PROT_CHUNKSIZE);
  connstat->commbufsiz = DEFCOMMBUFSIZ;
  connstat->tapebuf = NEWP(UChar, MPX_CHUNKSIZE);
  connstat->tapebufsiz = MPX_CHUNKSIZE; 
  connstat->req_streamermode = (nobuffering ? 0 : BUFFERED_OPERATION)
					| CHANGE_CART_ON_EOT;
  connstat->peeraddr = addr;
  connstat->peername = peername;
  set_client_id(connstat, connstat->peername);

  if(! connstat->peername || ! connstat->inbuf || ! connstat->outbuf
			|| ! connstat->clientid|| !connstat->tapebuf){
    ZFREE(connstat->peername);
    ZFREE(connstat->clientid);
    ZFREE(connstat->inbuf);
    ZFREE(connstat->outbuf);
    return(NULL);
  }

  connstat->inbuf_allocated = connstat->outbuf_allocated = MAX_PROT_CHUNKSIZE;
  connstat->req_cartset = 1;

  return(connstat);
}

static Int32
close_conn_func(
  int		fd,
  void		*conn_data,
  Int32		idx,
  void		*user_data,
  void		*tcpmux_status)
{
  ConnStatus	*connstat;

  connstat = *((ConnStatus **) conn_data);

  if(connstat->tape_status != NOT_OPEN){
    closetape(CLOSETAPE, fd, conn_data, idx, user_data, fromfd, tofd, tcpmux_status);
  }

  ZFREE(connstat->peername);
  ZFREE(connstat->clientid);
  ZFREE(connstat->inbuf);
  ZFREE(connstat->outbuf);
  ZFREE(connstat->tapebuf);

  /* handle buffer contents depending on conn_status ... */

  free(connstat);