text: /*
   * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions
   * are met:
   * 1. Redistributions of source code must retain the above copyright
   *    notice, this list of conditions and the following disclaimer.
   * 2. Redistributions in binary form must reproduce the above copyright
   *    notice, this list of conditions and the following disclaimer in the
   *    documentation and/or other materials provided with the distribution.
   * 3. The name of the author may not be used to endorse or promote products
   *    derived from this software without specific prior written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
   * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
   * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
   * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
   * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
   * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
   * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
   * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
   * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
   * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   */
  #include "util-internal.h"
  
  /* The old tests here need assertions to work. */
  #undef NDEBUG
  
  #include "event2/event-config.h"
  
  #include <sys/types.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
  #ifdef EVENT__HAVE_UNISTD_H
  #include <unistd.h>
  #endif
  #ifdef EVENT__HAVE_SYS_WAIT_H
  #include <sys/wait.h>
  #endif
  
  #ifdef EVENT__HAVE_PTHREADS
  #include <pthread.h>
  #elif defined(_WIN32)
  #include <process.h>
  #endif
  #include <assert.h>
  #ifdef EVENT__HAVE_UNISTD_H
  #include <unistd.h>
  #endif
  #include <time.h>
  
  #include "sys/queue.h"
  
  #include "event2/event.h"
  #include "event2/event_struct.h"
  #include "event2/thread.h"
  #include "event2/util.h"
  #include "evthread-internal.h"
  #include "event-internal.h"
  #include "defer-internal.h"
  #include "regress.h"
  #include "tinytest_macros.h"
  #include "time-internal.h"
  #include "regress_thread.h"
  
  /**
   * Put into the real time scheduling class for better timers latency.
   * https://developer.apple.com/library/archive/technotes/tn2169/_index.html#//apple_ref/doc/uid/DTS40013172-CH1-TNTAG6000
   */
  #if defined(__APPLE__)
  #ifdef EVENT__HAVE_MACH_MACH_H
  #include <mach/mach.h>
  #endif
  #ifdef EVENT__HAVE_MACH_MACH_TIME_H
  #include <mach/mach_time.h>
  #endif
  static void move_pthread_to_realtime_scheduling_class(pthread_t pthread)
  {
  	mach_timebase_info_data_t info;
  	mach_timebase_info(&info);
  
  	const uint64_t NANOS_PER_MSEC = 1000000ULL;
  	double clock2abs =
  		((double)info.denom / (double)info.numer) * NANOS_PER_MSEC;
  
  	thread_time_constraint_policy_data_t policy;
  	policy.period      = 0;
  	policy.computation = (uint32_t)(5 * clock2abs); // 5 ms of work
  	policy.constraint  = (uint32_t)(10 * clock2abs);
  	policy.preemptible = FALSE;
  
  	int kr = thread_policy_set(pthread_mach_thread_np(pthread),
  		THREAD_TIME_CONSTRAINT_POLICY,
  		(thread_policy_t)&policy,
  		THREAD_TIME_CONSTRAINT_POLICY_COUNT);
  	if (kr != KERN_SUCCESS) {
  		mach_error("thread_policy_set:", kr);
  		exit(1);
  	}
  }
  
  void thread_setup(THREAD_T pthread)
  {
  	move_pthread_to_realtime_scheduling_class(pthread);
  }
  #else /** \__APPLE__ */
  void thread_setup(THREAD_T pthread) {}
  #endif /** \!__APPLE__ */
  
  
  struct cond_wait {
  	void *lock;
  	void *cond;
  };
  
  static void
  wake_all_timeout(evutil_socket_t fd, short what, void *arg)
  {
  	struct cond_wait *cw = arg;
  	EVLOCK_LOCK(cw->lock, 0);
  	EVTHREAD_COND_BROADCAST(cw->cond);
  	EVLOCK_UNLOCK(cw->lock, 0);
  
  }
  
  static void
  wake_one_timeout(evutil_socket_t fd, short what, void *arg)
  {
  	struct cond_wait *cw = arg;
  	EVLOCK_LOCK(cw->lock, 0);
  	EVTHREAD_COND_SIGNAL(cw->cond);
  	EVLOCK_UNLOCK(cw->lock, 0);
  }
  
  #define NUM_THREADS	100
  #define NUM_ITERATIONS  100
  void *count_lock;
  static int count;
  
  static THREAD_FN
  basic_thread(void *arg)
  {
  	struct cond_wait cw;
  	struct event_base *base = arg;
  	struct event ev;
  	int i = 0;
  
  	EVTHREAD_ALLOC_LOCK(cw.lock, 0);
  	EVTHREAD_ALLOC_COND(cw.cond);
  	assert(cw.lock);
  	assert(cw.cond);
  
  	evtimer_assign(&ev, base, wake_all_timeout, &cw);
  	for (i = 0; i < NUM_ITERATIONS; i++) {
  		struct timeval tv;
  		evutil_timerclear(&tv);
  		tv.tv_sec = 0;
  		tv.tv_usec = 3000;
  
  		EVLOCK_LOCK(cw.lock, 0);
  		/* we need to make sure that event does not happen before
  		 * we get to wait on the conditional variable */
  		assert(evtimer_add(&ev, &tv) == 0);
  
  		assert(EVTHREAD_COND_WAIT(cw.cond, cw.lock) == 0);
  		EVLOCK_UNLOCK(cw.lock, 0);
  
  		EVLOCK_LOCK(count_lock, 0);
  		++count;
  		EVLOCK_UNLOCK(count_lock, 0);
  	}
  
  	/* exit the loop only if all threads fired all timeouts */
  	EVLOCK_LOCK(count_lock, 0);
  	if (count >= NUM_THREADS * NUM_ITERATIONS)
  		event_base_loopexit(base, NULL);
  	EVLOCK_UNLOCK(count_lock, 0);
  
  	EVTHREAD_FREE_LOCK(cw.lock, 0);
  	EVTHREAD_FREE_COND(cw.cond);
  
  	THREAD_RETURN();
  }
  
  static int notification_fd_used = 0;
  #ifndef _WIN32
  static int got_sigchld = 0;
  static void
  sigchld_cb(evutil_socket_t fd, short event, void *arg)
  {
  	struct timeval tv;
  	struct event_base *base = arg;
  
  	got_sigchld++;
  	tv.tv_usec = 100000;
  	tv.tv_sec = 0;
  	event_base_loopexit(base, &tv);
  }
  
  
  static void
  notify_fd_cb(evutil_socket_t fd, short event, void *arg)
  {
  	++notification_fd_used;
  }
  #endif
  
  static void
  thread_basic(void *arg)
  {
  	THREAD_T threads[NUM_THREADS];
  	struct event ev;
  	struct timeval tv;
  	int i;
  	struct basic_test_data *data = arg;
  	struct event_base *base = data->base;
  
  	struct event *notification_event = NULL;
  	struct event *sigchld_event = NULL;
  
  	EVTHREAD_ALLOC_LOCK(count_lock, 0);
  	tt_assert(count_lock);
  
  	tt_assert(base);
  	if (evthread_make_base_notifiable(base)<0) {
  		tt_abort_msg("Couldn't make base notifiable!");
  	}
  
  #ifndef _WIN32
  	if (data->setup_data && !strcmp(data->setup_data, "forking")) {
  		pid_t pid;
  		int status;
  		sigchld_event = evsignal_new(base, SIGCHLD, sigchld_cb, base);
  		/* This piggybacks on the th_notify_fd weirdly, and looks
  		 * inside libevent internals.  Not a good idea in non-testing
  		 * code! */
  		notification_event = event_new(base,
  		    base->th_notify_fd[0], EV_READ|EV_PERSIST, notify_fd_cb,
  		    NULL);
  		event_add(sigchld_event, NULL);
  		event_add(notification_event, NULL);
  
  		if ((pid = fork()) == 0) {
  			event_del(notification_event);
  			if (event_reinit(base) < 0) {
  				TT_FAIL(("reinit"));
  				exit(1);
  			}
  			event_assign(notification_event, base,
  			    base->th_notify_fd[0], EV_READ|EV_PERSIST,
  			    notify_fd_cb, NULL);
  			event_add(notification_event, NULL);
  	 		goto child;
  		}
  
  		event_base_dispatch(base);
  
  		if (waitpid(pid, &status, 0) == -1)
  			tt_abort_perror("waitpid");
  		TT_BLATHER(("Waitpid okay\n"));
  
  		tt_assert(got_sigchld);
  		tt_int_op(notification_fd_used, ==, 0);
  
  		goto end;
  	}
  
  child:
  #endif
  	for (i = 0; i < NUM_THREADS; ++i)
  		THREAD_START(threads[i], basic_thread, base);
  
  	evtimer_assign(&ev, base, NULL, NULL);
  	evutil_timerclear(&tv);
  	tv.tv_sec = 1000;
  	event_add(&ev, &tv);
  
  	event_base_dispatch(base);
  
  	for (i = 0; i < NUM_THREADS; ++i)
  		THREAD_JOIN(threads[i]);
  
  	event_del(&ev);
  
  	tt_int_op(count, ==, NUM_THREADS * NUM_ITERATIONS);
  
  	EVTHREAD_FREE_LOCK(count_lock, 0);
  
  	TT_BLATHER(("notifiations==%d", notification_fd_used));
  
  end:
  
  	if (notification_event)
  		event_free(notification_event);
  	if (sigchld_event)
  		event_free(sigchld_event);
  }
  
  #undef NUM_THREADS
  #define NUM_THREADS 10
  
  struct alerted_record {
  	struct cond_wait *cond;
  	struct timeval delay;
  	struct timeval alerted_at;
  	int timed_out;
  };
  
  static THREAD_FN
  wait_for_condition(void *arg)
  {
  	struct alerted_record *rec = arg;
  	int r;
  
  	EVLOCK_LOCK(rec->cond->lock, 0);
  	if (rec->delay.tv_sec || rec->delay.tv_usec) {
  		r = EVTHREAD_COND_WAIT_TIMED(rec->cond->cond, rec->cond->lock,
  		    &rec->delay);
  	} else {
  		r = EVTHREAD_COND_WAIT(rec->cond->cond, rec->cond->lock);
  	}
  	EVLOCK_UNLOCK(rec->cond->lock, 0);
  
  	evutil_gettimeofday(&rec->alerted_at, NULL);
  	if (r == 1)
  		rec->timed_out = 1;
  
  	THREAD_RETURN();
  }
  
  static void
  thread_conditions_simple(void *arg)
  {
  	struct timeval tv_signal, tv_timeout, tv_broadcast;
  	struct alerted_record alerted[NUM_THREADS];
  	THREAD_T threads[NUM_THREADS];
  	struct cond_wait cond;
  	int i;
  	struct timeval launched_at;
  	struct event wake_one;
  	struct event wake_all;
  	struct basic_test_data *data = arg;
  	struct event_base *base = data->base;
  	int n_timed_out=0, n_signal=0, n_broadcast=0;
  
  	tv_signal.tv_sec = tv_timeout.tv_sec = tv_broadcast.tv_sec = 0;
  	tv_signal.tv_usec = 30*1000;
  	tv_timeout.tv_usec = 150*1000;
  	tv_broadcast.tv_usec = 500*1000;
  
  	EVTHREAD_ALLOC_LOCK(cond.lock, EVTHREAD_LOCKTYPE_RECURSIVE);
  	EVTHREAD_ALLOC_COND(cond.cond);
  	tt_assert(cond.lock);
  	tt_assert(cond.cond);
  	for (i = 0; i < NUM_THREADS; ++i) {
  		memset(&alerted[i], 0, sizeof(struct alerted_record));
  		alerted[i].cond = &cond;
  	}
  
  	/* Threads 5 and 6 will be allowed to time out */
  	memcpy(&alerted[5].delay, &tv_timeout, sizeof(tv_timeout));
  	memcpy(&alerted[6].delay, &tv_timeout, sizeof(tv_timeout));
  
  	evtimer_assign(&wake_one, base, wake_one_timeout, &cond);
  	evtimer_assign(&wake_all, base, wake_all_timeout, &cond);
  
  	evutil_gettimeofday(&launched_at, NULL);
  
  	/* Launch the threads... */
  	for (i = 0; i < NUM_THREADS; ++i) {
  		THREAD_START(threads[i], wait_for_condition, &alerted[i]);
  	}
  
  	/* Start the timers... */
  	tt_int_op(event_add(&wake_one, &tv_signal), ==, 0);
  	tt_int_op(event_add(&wake_all, &tv_broadcast), ==, 0);
  
  	/* And run for a bit... */
  	event_base_dispatch(base);
  
  	/* And wait till the threads are done. */
  	for (i = 0; i < NUM_THREADS; ++i)
  		THREAD_JOIN(threads[i]);
  
  	/* Now, let's see what happened. At least one of 5 or 6 should
  	 * have timed out. */
  	n_timed_out = alerted[5].timed_out + alerted[6].timed_out;
  	tt_int_op(n_timed_out, >=, 1);
  	tt_int_op(n_timed_out, <=, 2);
  
  	for (i = 0; i < NUM_THREADS; ++i) {
  		const struct timeval *target_delay;
  		struct timeval target_time, actual_delay;
  		if (alerted[i].timed_out) {
  			TT_BLATHER(("%d looks like a timeout\n", i));
  			target_delay = &tv_timeout;
  			tt_assert(i == 5 || i == 6);
  		} else if (evutil_timerisset(&alerted[i].alerted_at)) {
  			long diff1,diff2;
  			evutil_timersub(&alerted[i].alerted_at,
  			    &launched_at, &actual_delay);
  			diff1 = timeval_msec_diff(&actual_delay,
  			    &tv_signal);
  			diff2 = timeval_msec_diff(&actual_delay,
  			    &tv_broadcast);
  			if (labs(diff1) < labs(diff2)) {
  				TT_BLATHER(("%d looks like a signal\n", i));
  				target_delay = &tv_signal;
  				++n_signal;
  			} else {
  				TT_BLATHER(("%d looks like a broadcast\n", i));
  				target_delay = &tv_broadcast;
  				++n_broadcast;
  			}
  		} else {
  			TT_FAIL(("Thread %d never got woken", i));
  			continue;
  		}
  		evutil_timeradd(target_delay, &launched_at, &target_time);
  		test_timeval_diff_leq(&target_time, &alerted[i].alerted_at,
  		    0, 200);
  	}
  	tt_int_op(n_broadcast + n_signal + n_timed_out, ==, NUM_THREADS);
  	tt_int_op(n_signal, ==, 1);
  
  end:
  	EVTHREAD_FREE_LOCK(cond.lock, EVTHREAD_LOCKTYPE_RECURSIVE);
  	EVTHREAD_FREE_COND(cond.cond);
  }
  
  #define CB_COUNT 128
  #define QUEUE_THREAD_COUNT 8
  
  static void
  SLEEP_MS(int ms)
  {
  	struct timeval tv;
  	tv.tv_sec = ms/1000;
  	tv.tv_usec = (ms%1000)*1000;
  	evutil_usleep_(&tv);
  }
  
  struct deferred_test_data {
  	struct event_callback cbs[CB_COUNT];
  	struct event_base *queue;
  };
  
  static struct timeval timer_start = {0,0};
  static struct timeval timer_end = {0,0};
  static unsigned callback_count = 0;
  static THREAD_T load_threads[QUEUE_THREAD_COUNT];
  static struct deferred_test_data deferred_data[QUEUE_THREAD_COUNT];
  
  static void
  deferred_callback(struct event_callback *cb, void *arg)
  {
  	SLEEP_MS(1);
  	callback_count += 1;
  }
  
  static THREAD_FN
  load_deferred_queue(void *arg)
  {
  	struct deferred_test_data *data = arg;
  	size_t i;
  
  	for (i = 0; i < CB_COUNT; ++i) {
  		event_deferred_cb_init_(&data->cbs[i], 0, deferred_callback,
  		    NULL);
  		event_deferred_cb_schedule_(data->queue, &data->cbs[i]);
  		SLEEP_MS(1);
  	}
  
  	THREAD_RETURN();
  }
  
  static void
  timer_callback(evutil_socket_t fd, short what, void *arg)
  {
  	evutil_gettimeofday(&timer_end, NULL);
  }
  
  static void
  start_threads_callback(evutil_socket_t fd, short what, void *arg)
  {
  	int i;
  
  	for (i = 0; i < QUEUE_THREAD_COUNT; ++i) {
  		THREAD_START(load_threads[i], load_deferred_queue,
  				&deferred_data[i]);
  	}
  }
  
  static void
  thread_deferred_cb_skew(void *arg)
  {
  	struct timeval tv_timer = {1, 0};
  	struct event_base *base = NULL;
  	struct event_config *cfg = NULL;
  	struct timeval elapsed;
  	int elapsed_usec;
  	int i;
  
  	cfg = event_config_new();
  	tt_assert(cfg);
  	event_config_set_max_dispatch_interval(cfg, NULL, 16, 0);
  
  	base = event_base_new_with_config(cfg);
  	tt_assert(base);
  
  	for (i = 0; i < QUEUE_THREAD_COUNT; ++i)
  		deferred_data[i].queue = base;
  
  	evutil_gettimeofday(&timer_start, NULL);
  	event_base_once(base, -1, EV_TIMEOUT, timer_callback, NULL,
  			&tv_timer);
  	event_base_once(base, -1, EV_TIMEOUT, start_threads_callback,
  			NULL, NULL);
  	event_base_dispatch(base);
  
  	evutil_timersub(&timer_end, &timer_start, &elapsed);
  	TT_BLATHER(("callback count, %u", callback_count));
  	elapsed_usec =
  	    (unsigned)(elapsed.tv_sec*1000000 + elapsed.tv_usec);
  	TT_BLATHER(("elapsed time, %u usec", elapsed_usec));
  
  	/* XXX be more intelligent here.  just make sure skew is
  	 * within .4 seconds for now. */
  	tt_assert(elapsed_usec >= 600000 && elapsed_usec <= 1400000);
  
  end:
  	for (i = 0; i < QUEUE_THREAD_COUNT; ++i)
  		THREAD_JOIN(load_threads[i]);
  	if (base)
  		event_base_free(base);
  	if (cfg)
  		event_config_free(cfg);
  }
  
  static struct event time_events[5];
  static struct timeval times[5];
  static struct event_base *exit_base = NULL;
  static void
  note_time_cb(evutil_socket_t fd, short what, void *arg)
  {
  	evutil_gettimeofday(arg, NULL);
  	if (arg == &times[4]) {
  		event_base_loopbreak(exit_base);
  	}
  }
  static THREAD_FN
  register_events_subthread(void *arg)
  {
  	struct timeval tv = {0,0};
  	SLEEP_MS(100);
  	event_active(&time_events[0], EV_TIMEOUT, 1);
  	SLEEP_MS(100);
  	event_active(&time_events[1], EV_TIMEOUT, 1);
  	SLEEP_MS(100);
  	tv.tv_usec = 100*1000;
  	event_add(&time_events[2], &tv);
  	tv.tv_usec = 150*1000;
  	event_add(&time_events[3], &tv);
  	SLEEP_MS(200);
  	event_active(&time_events[4], EV_TIMEOUT, 1);
  
  	THREAD_RETURN();
  }
  
  static void
  thread_no_events(void *arg)
  {
  	THREAD_T thread;
  	struct basic_test_data *data = arg;
  	struct timeval starttime, endtime;
  	int i;
  	exit_base = data->base;
  
  	memset(times,0,sizeof(times));
  	for (i=0;i<5;++i) {
  		event_assign(&time_events[i], data->base,
  		    -1, 0, note_time_cb, &times[i]);
  	}
  
  	evutil_gettimeofday(&starttime, NULL);
  	THREAD_START(thread, register_events_subthread, data->base);
  	event_base_loop(data->base, EVLOOP_NO_EXIT_ON_EMPTY);
  	evutil_gettimeofday(&endtime, NULL);
  	tt_assert(event_base_got_break(data->base));
  	THREAD_JOIN(thread);
  	for (i=0; i<5; ++i) {
  		struct timeval diff;
  		double sec;
  		evutil_timersub(&times[i], &starttime, &diff);
  		sec = diff.tv_sec + diff.tv_usec/1.0e6;
  		TT_BLATHER(("event %d at %.4f seconds", i, sec));
  	}
  	test_timeval_diff_eq(&starttime, &times[0], 100);
  	test_timeval_diff_eq(&starttime, &times[1], 200);
  	test_timeval_diff_eq(&starttime, &times[2], 400);
  	test_timeval_diff_eq(&starttime, &times[3], 450);
  	test_timeval_diff_eq(&starttime, &times[4], 500);
  	test_timeval_diff_eq(&starttime, &endtime,  500);
  
  end:
  	;
  }
  
  #define TEST(name, f)							\
  	{ #name, thread_##name, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE|(f),	\
  	  &basic_setup, NULL }
  
  struct testcase_t thread_testcases[] = {
  	{ "basic", thread_basic, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE,
  	  &basic_setup, NULL },
  #ifndef _WIN32
  	{ "forking", thread_basic, TT_FORK|TT_NEED_THREADS|TT_NEED_BASE,
  	  &basic_setup, (char*)"forking" },
  	{ "priority_inheritance", thread_basic,
  	  TT_FORK|TT_NEED_THREADS|TT_NEED_BASE|TT_ENABLE_PRIORITY_INHERITANCE,
  	  &basic_setup, (char*)"priority_inheritance" },
  #endif
  	TEST(conditions_simple, TT_RETRIABLE),
  	{ "deferred_cb_skew", thread_deferred_cb_skew,
  	  TT_FORK|TT_NEED_THREADS|TT_OFF_BY_DEFAULT,
  	  &basic_setup, NULL },
  #ifndef _WIN32
  	/****** XXX TODO FIXME windows seems to be having some timing trouble,
  	 * looking into it now. / ellzey
  	 ******/
  	TEST(no_events, TT_RETRIABLE),
  #endif
  	END_OF_TESTCASES
  };
  
  