rpc_server.c

这是一个开放源代码的与WINNT/WIN2K/WIN2003兼容的操作系统

#endif
}

static DWORD CALLBACK RPCRT4_worker_thread(LPVOID the_arg)
{
  DWORD obj;
  RpcPacket* pkt;

  for (;;) {
    /* idle timeout after 5s */
    obj = WaitForSingleObject(server_sem, 5000);
    if (obj == WAIT_TIMEOUT) {
      /* if another idle thread exist, self-destruct */
      if (worker_free > 1) break;
      continue;
    }
    pkt = RPCRT4_pop_packet();
    if (!pkt) continue;
    InterlockedDecrement(&worker_free);
    for (;;) {
      RPCRT4_process_packet(pkt->conn, pkt->hdr, pkt->msg);
      HeapFree(GetProcessHeap(), 0, pkt);
      /* try to grab another packet here without waiting
       * on the semaphore, in case it hits max */
      pkt = RPCRT4_pop_packet();
      if (!pkt) break;
      /* decrement semaphore */
      WaitForSingleObject(server_sem, 0);
    }
    InterlockedIncrement(&worker_free);
  }
  InterlockedDecrement(&worker_free);
  InterlockedDecrement(&worker_count);
  return 0;
}

static void RPCRT4_create_worker_if_needed(void)
{
  if (!worker_free && worker_count < MAX_THREADS) {
    HANDLE thread;
    InterlockedIncrement(&worker_count);
    InterlockedIncrement(&worker_free);
    thread = CreateThread(NULL, 0, RPCRT4_worker_thread, NULL, 0, NULL);
    if (thread) CloseHandle(thread);
    else {
      InterlockedDecrement(&worker_free);
      InterlockedDecrement(&worker_count);
    }
  }
}

static DWORD CALLBACK RPCRT4_io_thread(LPVOID the_arg)
{
  RpcConnection* conn = (RpcConnection*)the_arg;
  RpcPktHdr *hdr;
  RpcBinding *pbind;
  RPC_MESSAGE *msg;
  RPC_STATUS status;
  RpcPacket *packet;

  TRACE("(%p)\n", conn);

  for (;;) {
    msg = HeapAlloc(GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(RPC_MESSAGE));

    /* create temporary binding for dispatch, it will be freed in
     * RPCRT4_process_packet */
    RPCRT4_MakeBinding(&pbind, conn);
    msg->Handle = (RPC_BINDING_HANDLE)pbind;

    status = RPCRT4_Receive(conn, &hdr, msg);
    if (status != RPC_S_OK) {
      WARN("receive failed with error %lx\n", status);
      break;
    }

#if 0
    RPCRT4_process_packet(conn, hdr, msg);
#else
    packet = HeapAlloc(GetProcessHeap(), 0, sizeof(RpcPacket));
    packet->conn = conn;
    packet->hdr = hdr;
    packet->msg = msg;
    RPCRT4_create_worker_if_needed();
    RPCRT4_push_packet(packet);
    ReleaseSemaphore(server_sem, 1, NULL);
#endif
    msg = NULL;
  }
  HeapFree(GetProcessHeap(), 0, msg);
  RPCRT4_DestroyConnection(conn);
  return 0;
}

static void RPCRT4_new_client(RpcConnection* conn)
{
  HANDLE thread = CreateThread(NULL, 0, RPCRT4_io_thread, conn, 0, NULL);
  if (!thread) {
    DWORD err = GetLastError();
    ERR("failed to create thread, error=%08lx\n", err);
    RPCRT4_DestroyConnection(conn);
  }
  /* we could set conn->thread, but then we'd have to make the io_thread wait
   * for that, otherwise the thread might finish, destroy the connection, and
   * free the memory we'd write to before we did, causing crashes and stuff -
   * so let's implement that later, when we really need conn->thread */

  CloseHandle( thread );
}

static DWORD CALLBACK RPCRT4_server_thread(LPVOID the_arg)
{
  HANDLE m_event = mgr_event, b_handle;
  HANDLE *objs = NULL;
  DWORD count, res;
  RpcServerProtseq* cps;
  RpcConnection* conn;
  RpcConnection* cconn;
  BOOL set_ready_event = FALSE;

  TRACE("(the_arg == ^%p)\n", the_arg);

  for (;;) {
    EnterCriticalSection(&server_cs);
    /* open and count connections */
    count = 1;
    cps = protseqs;
    while (cps) {
      conn = cps->conn;
      while (conn) {
        RPCRT4_OpenConnection(conn);
        if (conn->ovl[0].hEvent) count++;
        conn = conn->Next;
      }
      cps = cps->Next;
    }
    /* make array of connections */
    if (objs)
	objs = HeapReAlloc(GetProcessHeap(), 0, objs, count*sizeof(HANDLE));
    else
	objs = HeapAlloc(GetProcessHeap(), 0, count*sizeof(HANDLE));

    objs[0] = m_event;
    count = 1;
    cps = protseqs;
    while (cps) {
      conn = cps->conn;
      while (conn) {
        if (conn->ovl[0].hEvent) objs[count++] = conn->ovl[0].hEvent;
        conn = conn->Next;
      }
      cps = cps->Next;
    }
    LeaveCriticalSection(&server_cs);

    if (set_ready_event)
    {
        /* signal to function that changed state that we are now sync'ed */
        SetEvent(server_ready_event);
        set_ready_event = FALSE;
    }

    /* start waiting */
    res = WaitForMultipleObjects(count, objs, FALSE, INFINITE);
    if (res == WAIT_OBJECT_0) {
      if (!std_listen)
      {
        SetEvent(server_ready_event);
        break;
      }
      set_ready_event = TRUE;
    }
    else if (res == WAIT_FAILED) {
      ERR("wait failed\n");
    }
    else {
      b_handle = objs[res - WAIT_OBJECT_0];
      /* find which connection got a RPC */
      EnterCriticalSection(&server_cs);
      conn = NULL;
      cps = protseqs;
      while (cps) {
        conn = cps->conn;
        while (conn) {
          if (conn->ovl[0].hEvent == b_handle) break;
          conn = conn->Next;
        }
        if (conn) break;
        cps = cps->Next;
      }
      cconn = NULL;
      if (conn) RPCRT4_SpawnConnection(&cconn, conn);
      LeaveCriticalSection(&server_cs);
      if (!conn) {
        ERR("failed to locate connection for handle %p\n", b_handle);
      }
      if (cconn) RPCRT4_new_client(cconn);
    }
  }
  HeapFree(GetProcessHeap(), 0, objs);
  EnterCriticalSection(&server_cs);
  /* close connections */
  cps = protseqs;
  while (cps) {
    conn = cps->conn;
    while (conn) {
      RPCRT4_CloseConnection(conn);
      conn = conn->Next;
    }
    cps = cps->Next;
  }
  LeaveCriticalSection(&server_cs);
  return 0;
}

/* tells the server thread that the state has changed and waits for it to
 * make the changes */
static void RPCRT4_sync_with_server_thread(void)
{
  /* make sure we are the only thread sync'ing the server state, otherwise
   * there is a race with the server thread setting an older state and setting
   * the server_ready_event when the new state hasn't yet been applied */
  WaitForSingleObject(mgr_mutex, INFINITE);

  SetEvent(mgr_event);
  /* wait for server thread to make the requested changes before returning */
  WaitForSingleObject(server_ready_event, INFINITE);

  ReleaseMutex(mgr_mutex);
}

static RPC_STATUS RPCRT4_start_listen(BOOL auto_listen)
{
  RPC_STATUS status = RPC_S_ALREADY_LISTENING;

  TRACE("\n");

  EnterCriticalSection(&listen_cs);
  if (auto_listen || (manual_listen_count++ == 0))
  {
    status = RPC_S_OK;
    if (++listen_count == 1) {
      HANDLE server_thread;
      /* first listener creates server thread */
      if (!mgr_mutex) mgr_mutex = CreateMutexW(NULL, FALSE, NULL);
      if (!mgr_event) mgr_event = CreateEventW(NULL, FALSE, FALSE, NULL);
      if (!server_ready_event) server_ready_event = CreateEventW(NULL, FALSE, FALSE, NULL);
      if (!server_sem) server_sem = CreateSemaphoreW(NULL, 0, MAX_THREADS, NULL);
      if (!worker_tls) worker_tls = TlsAlloc();
      std_listen = TRUE;
      server_thread = CreateThread(NULL, 0, RPCRT4_server_thread, NULL, 0, NULL);
      CloseHandle(server_thread);
    }
  }
  LeaveCriticalSection(&listen_cs);

  return status;
}

static void RPCRT4_stop_listen(BOOL auto_listen)
{
  EnterCriticalSection(&listen_cs);
  if (auto_listen || (--manual_listen_count == 0))
  {
    if (listen_count != 0 && --listen_count == 0) {
      std_listen = FALSE;
      LeaveCriticalSection(&listen_cs);
      RPCRT4_sync_with_server_thread();
      return;
    }
    assert(listen_count >= 0);
  }
  LeaveCriticalSection(&listen_cs);
}

static RPC_STATUS RPCRT4_use_protseq(RpcServerProtseq* ps)
{
  RPCRT4_CreateConnection(&ps->conn, TRUE, ps->Protseq, NULL, ps->Endpoint, NULL, NULL);

  EnterCriticalSection(&server_cs);
  ps->Next = protseqs;
  protseqs = ps;
  LeaveCriticalSection(&server_cs);

  if (std_listen) RPCRT4_sync_with_server_thread();

  return RPC_S_OK;
}

/***********************************************************************
 *             RpcServerInqBindings (RPCRT4.@)
 */
RPC_STATUS WINAPI RpcServerInqBindings( RPC_BINDING_VECTOR** BindingVector )
{
  RPC_STATUS status;
  DWORD count;
  RpcServerProtseq* ps;
  RpcConnection* conn;

  if (BindingVector)
    TRACE("(*BindingVector == ^%p)\n", *BindingVector);
  else
    ERR("(BindingVector == NULL!!?)\n");

  EnterCriticalSection(&server_cs);
  /* count connections */
  count = 0;
  ps = protseqs;
  while (ps) {
    conn = ps->conn;
    while (conn) {
      count++;
      conn = conn->Next;
    }
    ps = ps->Next;
  }
  if (count) {
    /* export bindings */
    *BindingVector = HeapAlloc(GetProcessHeap(), 0,
                              sizeof(RPC_BINDING_VECTOR) +
                              sizeof(RPC_BINDING_HANDLE)*(count-1));
    (*BindingVector)->Count = count;
    count = 0;
    ps = protseqs;
    while (ps) {
      conn = ps->conn;
      while (conn) {
       RPCRT4_MakeBinding((RpcBinding**)&(*BindingVector)->BindingH[count],
                          conn);
       count++;
       conn = conn->Next;
      }
      ps = ps->Next;
    }
    status = RPC_S_OK;
  } else {
    *BindingVector = NULL;
    status = RPC_S_NO_BINDINGS;
  }
  LeaveCriticalSection(&server_cs);
  return status;
}

/***********************************************************************
 *             RpcServerUseProtseqEpA (RPCRT4.@)
 */
RPC_STATUS WINAPI RpcServerUseProtseqEpA( unsigned char *Protseq, UINT MaxCalls, unsigned char *Endpoint, LPVOID SecurityDescriptor )
{
  RPC_POLICY policy;
  
  TRACE( "(%s,%u,%s,%p)\n", Protseq, MaxCalls, Endpoint, SecurityDescriptor );
  
  /* This should provide the default behaviour */
  policy.Length        = sizeof( policy );
  policy.EndpointFlags = 0;