async_io.c

用c++包装好的线程库,直接拿来使用,提高效率.

//
// This file is part of the C++ Threads library.
//
// Copyright (C) 2001 Orn E. Hansen
//

#include "io.h"
#include "signal_num.h"
#include "sig_handler.h"

#include <iostream>

#include <cstdio>
#include <csignal>

namespace cpp_threads {

  typedef signalling<AsynchronousIO>    async_signal_t;

  struct parameter_class {
    bool           s_data;
    int            s_invoker;
    SocketBuffer*  s_sbuffer;
  };

  AsynchronousIO::AsynchronousIO(async_type_t p_type)
  {
    Pthread::debug("AsynchronousIO::AsynchronousIO");
    kind(Mutex::fast_e);
    _mode   = p_type;
    _client = 0;
    _ssyst  = new async_signal_t;
    if ( _mode == dupl_deliver_e )
      ((async_signal_t *)_ssyst)->connect(s_io_ready,this,&AsynchronousIO::signalIOReady);
    ((async_signal_t *)_ssyst)->connect(s_urgent,this,&AsynchronousIO::signalUrgent);
    _pid  = getpid();
  }

  AsynchronousIO::~AsynchronousIO()
  {
    Pthread::debug("AsynchronousIO::~AsynchronousIO");
  }

  SocketBuffer&
  AsynchronousIO::fd()
  {
    return _fd;
  }

  //
  // This is a new version of the signal handling system, the
  // signal is now delivered to the following function, and along
  // with it parameters needed.  The class itself is already
  // locked when this entry is taken, so all mechanisms to lock
  // this entry is no longer necessary.
  void
  AsynchronousIO::signalIOReady(void *param_p)
  {
    parameter_class *param = (parameter_class *)param_p;
    Pthread*              th = Pthread::self();
    Pthread::thread_state saved;

    saved = th->state();
    if( param ) {
      if( param->s_data ) {
	got(*param->s_sbuffer);
	kill(param->s_invoker,s_urgent);
      } else
	exception(*param->s_sbuffer);
      // Parameters were delivered in a temporary buffer.
      delete param;
    }
    th->state(saved);
    th->signal(s_io_ready);
  }

  void
  AsynchronousIO::signalUrgent(void *)
  {
    Pthread::debug("signalling the void");
  }

  void
  AsynchronousIO::exception(SocketBuffer&)
  {
  }

  void
  AsynchronousIO::queueSocket(SocketBuffer*)
  {
  }

  void
  AsynchronousIO::close(Socket&)
  {
  }

  //
  // jump
  //
  // Make an asynchronous jump to the thread that wants
  // this data handled.
  //
  void
  AsynchronousIO::jump(SocketBuffer* p_bsno,int p_len)
  {
    parameter_class *param;
    Pthread::debug("jump(%d)",p_len);

    if( getpid() != _pid ) {
      if( p_len > 0 )
	if( tryLock() ) {
	  if( p_bsno != _client )
	    queueSocket(p_bsno);
	  return;
	}
      param = new struct parameter_class;
      param->s_sbuffer = p_bsno ? p_bsno : &_fd;
      param->s_invoker = getpid();
      param->s_data    = p_len > 0;
      ((async_signal_t *)_ssyst)->emit(s_io_ready,(void *)param);
    } else {
      if( p_len > 0 )
	got( (p_bsno?*p_bsno:fd()) );
      else
	exception( (p_bsno?*p_bsno:fd()) );
    }
  }

  SocketBuffer&
  AsynchronousIO::clientFD()
  {
    if( _client )
      return *_client;
    return fd();
  }

}; // Namespace