int Database::flush() {
	int drainCount = 0;

	LOG4CPLUS_INFO(logger, "Flushing lists...");

	drainCount = drain();
	drainCount += drain();

	return drainCount;
}

 void GridTransceiver_Test::Test_DataRates() {
   GridTransceiver pt;
   for (u32 i = 0; i < 100; ++i) {
     if (i < 80 || (i%4 == 0)) {
       pt.Write(true, (const u8*) "abcdefghijklmnopqrstuvwxyz0123456789!!!!!!!!!!",strlen("abcdefghijklmnopqrstuvwxyz0123456789!!!!!!!!!!"));
       pt.Write(false,(const u8*) "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",strlen("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"));
     }
     pt.Advance(333333); // 1/3 million nanos = 1/3 ms
     drain(pt,true,i);
     drain(pt,false,i);
   }
 }

static void pulse_close(void) {

  ENTER(__FUNCTION__);
    
  drain();

  connected = 0;

  if (mainloop)
    pa_threaded_mainloop_stop(mainloop);

    connected = 0;

  if (context) {
    SHOW_TIME("pa_context_disconnect (call)");
    pa_context_disconnect(context);
    pa_context_unref(context);
    context = NULL;
  }
  
  if (mainloop) {
  SHOW_TIME("pa_threaded_mainloop_free (call)");
    pa_threaded_mainloop_free(mainloop);
    mainloop = NULL;
  }
  SHOW_TIME("pulse_close (ret)");
  
}

static void wait_drain(struct ao *ao)
{
    struct priv *priv = ao->priv;
    drain(ao);
    if (!priv->paused)
        mp_sleep_us(1000000.0 * priv->buffered / ao->samplerate / priv->speed);
}

static void wakeup(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
  GPR_TIMER_MARK("workqueue.wakeup", 0);
  grpc_error *err = grpc_wakeup_fd_wakeup(&workqueue->wakeup_fd);
  if (!GRPC_LOG_IF_ERROR("wakeupfd_wakeup", err)) {
    drain(exec_ctx, workqueue);
  }
}

/*
 * Release the specified connection.
 * Certain actions may be performed before doing this:
 *      - drain of a single UDP packet if the socket type is SOCK_DGRAM
 */
void conn_free( connection_s *cp, int release_mem )
{
   struct service *sp ;

   if( cp == NULL )
      return;
      if( debug.on )
         msg( LOG_INFO, "conn_free", "freeing connection") ;

   sp = cp->co_sp ;

   if( (SVC_SOCKET_TYPE( sp ) == SOCK_DGRAM) && (SVC_IS_ACTIVE( sp )) )
      drain( cp->co_descriptor ) ;

   if ( SVC_RELE( sp ) == 0 ) {
      pset_remove( SERVICES( ps ), sp ) ;
      svc_release( sp );
   }
   cp->co_sp = NULL;

   if ( CONN_DESCRIPTOR( cp ) > 0 )
      CONN_CLOSE( cp ) ;

   CLEAR( *cp ) ;
   if (release_mem) {
      FREE_CONN( cp ) ;
   }
}

int wave_close(void* theHandler)
{
  SHOW_TIME("wave_close > ENTER");
  static int aStopStreamCount = 0;

	// Avoid race condition by making sure this function only
	// gets called once at a time
	aStopStreamCount++;
	if (aStopStreamCount != 1)
	{
		SHOW_TIME("wave_close > LEAVE (stopStreamCount)");
		return 0;
	}

	int a_status = pthread_mutex_lock(&pulse_mutex);
	if (a_status)
	{
		SHOW("Error: pulse_mutex lock=%d (%s)\n", a_status, __FUNCTION__);
		aStopStreamCount = 0; // last action
		return PULSE_ERROR;
	}

	drain();

	pthread_mutex_unlock(&pulse_mutex);
	SHOW_TIME("wave_close (ret)");

	aStopStreamCount = 0; // last action
	return PULSE_OK;
}

			void flush()
			{
				if ( ! deflatecontext.deflateoutflushed )
				{
					drain();

					deflatecontext.deflategloblist.terminate();

					{
						deflatecontext.deflateexlock.lock();

						if ( deflatecontext.deflateexceptionid != std::numeric_limits<uint64_t>::max() )
						{
							deflatecontext.deflateexlock.unlock();

							libmaus2::parallel::ScopePosixMutex Q(deflatecontext.deflateexlock);
							throw (*(deflatecontext.deflatepse));
						}
						else
						{
							deflatecontext.deflateexlock.unlock();
						}
					}

					// write default compressed block with size 0 (EOF marker)
					libmaus2::lz::BgzfDeflateBase eofBase;
					BgzfDeflateZStreamBaseFlushInfo const eofflushsize = eofBase.flush(true /* full flush */);
					assert ( ! eofflushsize.movesize );
					deflatecontext.streamWrite(eofBase.inbuf.begin(),eofBase.outbuf.begin(),eofflushsize);

					deflatecontext.deflateoutflushed = true;
				}
			}

static double get_delay(struct ao *ao)
{
    struct priv *priv = ao->priv;

    drain(ao);

    // Note how get_delay returns the delay in audio device time (instead of
    // adjusting for speed), since most AOs seem to also do that.
    double delay = priv->buffered;

    // Drivers with broken EOF handling usually always report the same device-
    // level delay that is additional to the buffer time.
    if (priv->broken_eof && priv->buffered < priv->latency)
        delay = priv->latency;

    delay /= ao->samplerate;

    if (priv->broken_delay) { // Report only multiples of outburst
        double q = priv->outburst / (double)ao->samplerate;
        if (delay > 0)
            delay = (int)(delay / q) * q;
    }

    return delay;
}

// stop playing, keep buffers (for pause)
static void pause(struct ao *ao)
{
    struct priv *priv = ao->priv;

    drain(ao);
    priv->paused = true;
}

static
int finish(struct audio_finish *finish)
{
  int i, result = 0;

  if (opened) {
    if (drain() == -1)
      result = -1;
    if (close_dev(wave_handle) == -1)
      result = -1;
  }

  /* restore priority status */

  SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_NORMAL);
  SetPriorityClass(GetCurrentProcess(), NORMAL_PRIORITY_CLASS);

  for (i = 0; i < NBUFFERS; ++i) {
    if (CloseHandle(output[i].event_handle) == 0 && result == 0) {
      audio_error = _("failed to close synchronization object");
      result = -1;
    }
  }

  return result;
}

 wchar_t dereference_impl() {
     if(! m_full){
         m_current_value = drain();
         m_full = true;
     }
     return m_current_value;
 }

void di_exit(void)
{
   struct intercept_s *ip = &dgram_intercept_state ;

   if ( IDP( ip->int_priv )->received_packets == 0 )
      drain( INT_REMOTE( ip ) ) ;
   int_exit( ip ) ;
}

// resume playing, after pause()
static void resume(struct ao *ao)
{
    struct priv *priv = ao->priv;

    drain(ao);
    priv->paused = false;
    priv->last_time = mp_time_sec();
}

static int get_space(struct ao *ao)
{
    struct priv *priv = ao->priv;

    drain(ao);
    int samples = priv->buffersize - priv->latency - priv->buffered;
    return samples / priv->outburst * priv->outburst;
}

void svc_request( struct service *sp )
{
   connection_s *cp ;
   status_e ret_code;

   cp = conn_new( sp ) ;
   if ( cp == CONN_NULL )
      return ;

   /*
    * Output the banner now that the connection is established. The
    * other banners come later.
    */
   banner_always(sp, cp);

   if (SVC_NOT_GENERIC(sp))
      ret_code = spec_service_handler(sp, cp);
   else 
      ret_code = svc_generic_handler(sp, cp);

   if( (SVC_SOCKET_TYPE( sp ) == SOCK_DGRAM) && (SVC_IS_ACTIVE( sp )) ) 
      drain( cp->co_descriptor ) ; /* Prevents looping next time */
   
   if ( ret_code != OK ) 
   {
      if ( SVC_LOGS_USERID_ON_FAILURE( sp ) ) {
         if( spec_service_handler( LOG_SERVICE( ps ), cp ) == FAILED ) 
	    conn_free( cp, 1 ) ;
         else if (!SC_WAITS( SVC_CONF( sp ) ) ) {
	 /* The logging service will gen SIGCHLD thus freeing connection */
	    CONN_CLOSE(cp) ; 
	 }
	 return;
      }
      if (!SC_WAITS( SVC_CONF( sp ) )) 
	 conn_free( cp, 1 );
      else { 
         if( (SVC_SOCKET_TYPE( sp ) == SOCK_DGRAM) && (SVC_IS_ACTIVE( sp )) ) 
            drain( cp->co_descriptor ) ; /* Prevents looping next time */
	 free( cp );
      }
   }
   else if ((SVC_NOT_GENERIC(sp)) || (!SC_FORKS( SVC_CONF( sp ) ) ) )
     free( cp );
}

int net::Buffer::remove(void* data,size_t datlen)
{
	size_t nread=datlen;
	if (nread>=off_)
		nread=off_;
	memcpy(data,buffer_,nread);
	drain(nread);
	return (nread);
}

			void write(char const * c, uint64_t n)
			{
				while ( n )
				{
					uint64_t const freespace = deflatecontext.deflateB[deflatecontext.deflatecurobject]->pe - deflatecontext.deflateB[deflatecontext.deflatecurobject]->pc;
					uint64_t const towrite = std::min(n,freespace);
					std::copy(reinterpret_cast<uint8_t const *>(c),reinterpret_cast<uint8_t const *>(c)+towrite,deflatecontext.deflateB[deflatecontext.deflatecurobject]->pc);

					c += towrite;
					deflatecontext.deflateB[deflatecontext.deflatecurobject]->pc += towrite;
					n -= towrite;

					// if block is now full
					if ( deflatecontext.deflateB[deflatecontext.deflatecurobject]->pc == deflatecontext.deflateB[deflatecontext.deflatecurobject]->pe )
					{
						// check for exceptions on output
						{
							deflatecontext.deflateexlock.lock();

							if ( deflatecontext.deflateexceptionid != std::numeric_limits<uint64_t>::max() )
							{
								deflatecontext.deflateexlock.unlock();

								drain();

								libmaus2::parallel::ScopePosixMutex Q(deflatecontext.deflateexlock);
								throw (*(deflatecontext.deflatepse));
							}
							else
							{
								deflatecontext.deflateexlock.unlock();
							}
						}

						// push data object id into deflate queue
						{
							libmaus2::parallel::ScopePosixMutex Q(deflatecontext.deflateqlock);
							deflatecontext.deflatecompqueue.push_back(deflatecontext.deflatecurobject);
						}

						// register task in global todo list
						deflatecontext.deflategloblist.enque(
							BgzfThreadQueueElement(
								BgzfThreadOpBase::libmaus2_lz_bgzf_op_compress_block,
								deflatecontext.deflatecurobject,
								0 /* block id */
							)
						);

						// get next object
						deflatecontext.deflatecurobject = deflatecontext.deflatefreelist.deque();
						// set block id of next object
						deflatecontext.deflateB[deflatecontext.deflatecurobject]->blockid = deflatecontext.deflateoutid++;
					}
				}
			}

bool DrainLife::Zap(Coord dir)
{
    //Take away 1/2 of hero's hit points, then take it away evenly from the monsters in the room (or next to hero if he is in a passage)
    if (game->hero().get_hp() < 2) {
        msg("you are too weak to use it");
        return false;
    }
    drain();
    return true;
}

void run(int i, char *f) {
	pid = fork();
	if (pid < 0) {
		perror("Fork failed.");
		exit(201);
	} else if (pid == 0) {
		close(0);
		dup2(fds[0], 1);
		dup2(fds[0], 2);
		execlp("bash", "bash", f, NULL);
		perror("execlp");
		fflush(stderr);
		_exit(202);
	} else {
		char buf[128];
		snprintf(buf, 128, "%s ...", f);
		buf[127] = 0;
		printf("Running %-40s ", buf);
		fflush(stdout);
		int st, w;
		while ((w = waitpid(pid, &st, WNOHANG)) == 0) {
			drain();
			usleep(20000);
		}
		if (w != pid) {
			perror("waitpid");
			exit(206);
		}
		drain();
		if (WIFEXITED(st)) {
			if (WEXITSTATUS(st) == 0) {
				passed(i, f);
			} else if (WEXITSTATUS(st) == 200) {
				skipped(i, f);
			} else {
				failed(i, f, st);
			}
		} else {
			failed(i, f, st);
		}
		clear();
	}
}