[ViewVC] Diff of: cvs/libeio/eio.c

Comparing libeio/eio.c (file contents):
Revision 1.128 by root, Mon Feb 18 03:16:30 2013 UTC vs.
Revision 1.129 by root, Sun Apr 14 09:43:19 2013 UTC

…		…
1	/*	1	/*
2	* libeio implementation	2	* libeio implementation
3	*	3	*
4	* Copyright (c) 2007,2008,2009,2010,2011,2012 Marc Alexander Lehmann <libeio@schmorp.de>	4	* Copyright (c) 2007,2008,2009,2010,2011,2012,2013 Marc Alexander Lehmann <libeio@schmorp.de>
5	* All rights reserved.	5	* All rights reserved.
6	*	6	*
7	* Redistribution and use in source and binary forms, with or without modifica-	7	* Redistribution and use in source and binary forms, with or without modifica-
8	* tion, are permitted provided that the following conditions are met:	8	* tion, are permitted provided that the following conditions are met:
9	*	9	*
213	#endif	213	#endif
214		214
215	return EIO_ERRNO (ENOENT, -1);	215	return EIO_ERRNO (ENOENT, -1);
216	}	216	}
217		217
218	/* POSIX API only */	218	/* POSIX API only, causing trouble for win32 apps */
219	#define CreateHardLink(neu,old,flags) 0	219	#define CreateHardLink(neu,old,flags) 0 /* not really creating hardlink, still using relative paths? */
220	#define CreateSymbolicLink(neu,old,flags) 0	220	#define CreateSymbolicLink(neu,old,flags) 0 /* vista+ only */
221		221
222	struct statvfs	222	struct statvfs
223	{	223	{
224	int dummy;	224	int dummy;
225	};	225	};
…		…
333	return -1	333	return -1
334		334
335	#define FUBd \	335	#define FUBd \
336	free (eio_buf)	336	free (eio_buf)
337		337
338	#define EIO_TICKS ((1000000 + 1023) >> 10)
339
340	/*****************************************************************************/	338	/*****************************************************************************/
341		339
342	struct tmpbuf	340	struct tmpbuf
343	{	341	{
344	void *ptr;	342	void *ptr;
…		…
382	/*****************************************************************************/	380	/*****************************************************************************/
383		381
384	#define ETP_PRI_MIN EIO_PRI_MIN	382	#define ETP_PRI_MIN EIO_PRI_MIN
385	#define ETP_PRI_MAX EIO_PRI_MAX	383	#define ETP_PRI_MAX EIO_PRI_MAX
386		384
		385	#define ETP_TYPE_QUIT -1
		386	#define ETP_TYPE_GROUP EIO_GROUP
		387
387	struct etp_worker;	388	struct etp_worker;
388		389
389	#define ETP_REQ eio_req	390	#define ETP_REQ eio_req
390	#define ETP_DESTROY(req) eio_destroy (req)	391	#define ETP_DESTROY(req) eio_destroy (req)
391	static int eio_finish (eio_req *req);	392	static int eio_finish (eio_req *req);
392	#define ETP_FINISH(req) eio_finish (req)	393	#define ETP_FINISH(req) eio_finish (req)
393	static void eio_execute (struct etp_worker self, eio_req req);	394	static void eio_execute (struct etp_worker self, eio_req req);
394	#define ETP_EXECUTE(wrk,req) eio_execute (wrk,req)	395	#define ETP_EXECUTE(wrk,req) eio_execute (wrk, req)
395		396
396	/*****************************************************************************/	397	#include "etp.c"
397
398	#define ETP_NUM_PRI (ETP_PRI_MAX - ETP_PRI_MIN + 1)
399
400	/* calculate time difference in ~1/EIO_TICKS of a second */
401	ecb_inline int
402	tvdiff (struct timeval tv1, struct timeval tv2)
403	{
404	return (tv2->tv_sec - tv1->tv_sec ) * EIO_TICKS
405	+ ((tv2->tv_usec - tv1->tv_usec) >> 10);
406	}
407
408	static unsigned int started, idle, wanted = 4;
409
410	static void (*want_poll_cb) (void);
411	static void (*done_poll_cb) (void);
412
413	static unsigned int max_poll_time; /* reslock */
414	static unsigned int max_poll_reqs; /* reslock */
415
416	static unsigned int nreqs; /* reqlock */
417	static unsigned int nready; /* reqlock */
418	static unsigned int npending; /* reqlock */
419	static unsigned int max_idle = 4; /* maximum number of threads that can idle indefinitely */
420	static unsigned int idle_timeout = 10; /* number of seconds after which an idle threads exit */
421
422	static xmutex_t wrklock;
423	static xmutex_t reslock;
424	static xmutex_t reqlock;
425	static xcond_t reqwait;
426
427	typedef struct etp_worker
428	{
429	struct tmpbuf tmpbuf;
430
431	/* locked by wrklock */
432	struct etp_worker prev, next;
433
434	xthread_t tid;
435
436	#ifdef ETP_WORKER_COMMON
437	ETP_WORKER_COMMON
438	#endif
439	} etp_worker;
440
441	static etp_worker wrk_first; /* NOT etp */
442
443	#define ETP_WORKER_LOCK(wrk) X_LOCK (wrklock)
444	#define ETP_WORKER_UNLOCK(wrk) X_UNLOCK (wrklock)
445
446	/* worker threads management */
447
448	static void
449	etp_worker_clear (etp_worker *wrk)
450	{
451	}
452
453	static void ecb_cold
454	etp_worker_free (etp_worker *wrk)
455	{
456	free (wrk->tmpbuf.ptr);
457
458	wrk->next->prev = wrk->prev;
459	wrk->prev->next = wrk->next;
460
461	free (wrk);
462	}
463
464	static unsigned int
465	etp_nreqs (void)
466	{
467	int retval;
468	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
469	retval = nreqs;
470	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
471	return retval;
472	}
473
474	static unsigned int
475	etp_nready (void)
476	{
477	unsigned int retval;
478
479	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
480	retval = nready;
481	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
482
483	return retval;
484	}
485
486	static unsigned int
487	etp_npending (void)
488	{
489	unsigned int retval;
490
491	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
492	retval = npending;
493	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
494
495	return retval;
496	}
497
498	static unsigned int
499	etp_nthreads (void)
500	{
501	unsigned int retval;
502
503	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
504	retval = started;
505	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
506
507	return retval;
508	}
509
510	/*
511	* a somewhat faster data structure might be nice, but
512	* with 8 priorities this actually needs <20 insns
513	* per shift, the most expensive operation.
514	*/
515	typedef struct {
516	ETP_REQ qs[ETP_NUM_PRI], qe[ETP_NUM_PRI]; /* qstart, qend */
517	int size;
518	} etp_reqq;
519
520	static etp_reqq req_queue;
521	static etp_reqq res_queue;
522
523	static void ecb_noinline ecb_cold
524	reqq_init (etp_reqq *q)
525	{
526	int pri;
527
528	for (pri = 0; pri < ETP_NUM_PRI; ++pri)
529	q->qs[pri] = q->qe[pri] = 0;
530
531	q->size = 0;
532	}
533
534	static int ecb_noinline
535	reqq_push (etp_reqq q, ETP_REQ req)
536	{
537	int pri = req->pri;
538	req->next = 0;
539
540	if (q->qe[pri])
541	{
542	q->qe[pri]->next = req;
543	q->qe[pri] = req;
544	}
545	else
546	q->qe[pri] = q->qs[pri] = req;
547
548	return q->size++;
549	}
550
551	static ETP_REQ * ecb_noinline
552	reqq_shift (etp_reqq *q)
553	{
554	int pri;
555
556	if (!q->size)
557	return 0;
558
559	--q->size;
560
561	for (pri = ETP_NUM_PRI; pri--; )
562	{
563	eio_req *req = q->qs[pri];
564
565	if (req)
566	{
567	if (!(q->qs[pri] = (eio_req *)req->next))
568	q->qe[pri] = 0;
569
570	return req;
571	}
572	}
573
574	abort ();
575	}
576
577	static int ecb_cold
578	etp_init (void (want_poll)(void), void (done_poll)(void))
579	{
580	X_MUTEX_CREATE (wrklock);
581	X_MUTEX_CREATE (reslock);
582	X_MUTEX_CREATE (reqlock);
583	X_COND_CREATE (reqwait);
584
585	reqq_init (&req_queue);
586	reqq_init (&res_queue);
587
588	wrk_first.next =
589	wrk_first.prev = &wrk_first;
590
591	started = 0;
592	idle = 0;
593	nreqs = 0;
594	nready = 0;
595	npending = 0;
596
597	want_poll_cb = want_poll;
598	done_poll_cb = done_poll;
599
600	return 0;
601	}
602
603	X_THREAD_PROC (etp_proc);
604
605	static void ecb_cold
606	etp_start_thread (void)
607	{
608	etp_worker *wrk = calloc (1, sizeof (etp_worker));
609
610	/TODO/
611	assert (("unable to allocate worker thread data", wrk));
612
613	X_LOCK (wrklock);
614
615	if (xthread_create (&wrk->tid, etp_proc, (void *)wrk))
616	{
617	wrk->prev = &wrk_first;
618	wrk->next = wrk_first.next;
619	wrk_first.next->prev = wrk;
620	wrk_first.next = wrk;
621	++started;
622	}
623	else
624	free (wrk);
625
626	X_UNLOCK (wrklock);
627	}
628
629	static void
630	etp_maybe_start_thread (void)
631	{
632	if (ecb_expect_true (etp_nthreads () >= wanted))
633	return;
634
635	/* todo: maybe use idle here, but might be less exact */
636	if (ecb_expect_true (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()))
637	return;
638
639	etp_start_thread ();
640	}
641
642	static void ecb_cold
643	etp_end_thread (void)
644	{
645	eio_req req = calloc (1, sizeof (eio_req)); / will be freed by worker */
646
647	req->type = -1;
648	req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
649
650	X_LOCK (reqlock);
651	reqq_push (&req_queue, req);
652	X_COND_SIGNAL (reqwait);
653	X_UNLOCK (reqlock);
654
655	X_LOCK (wrklock);
656	--started;
657	X_UNLOCK (wrklock);
658	}
659
660	static int
661	etp_poll (void)
662	{
663	unsigned int maxreqs;
664	unsigned int maxtime;
665	struct timeval tv_start, tv_now;
666
667	X_LOCK (reslock);
668	maxreqs = max_poll_reqs;
669	maxtime = max_poll_time;
670	X_UNLOCK (reslock);
671
672	if (maxtime)
673	gettimeofday (&tv_start, 0);
674
675	for (;;)
676	{
677	ETP_REQ *req;
678
679	etp_maybe_start_thread ();
680
681	X_LOCK (reslock);
682	req = reqq_shift (&res_queue);
683
684	if (req)
685	{
686	--npending;
687
688	if (!res_queue.size && done_poll_cb)
689	done_poll_cb ();
690	}
691
692	X_UNLOCK (reslock);
693
694	if (!req)
695	return 0;
696
697	X_LOCK (reqlock);
698	--nreqs;
699	X_UNLOCK (reqlock);
700
701	if (ecb_expect_false (req->type == EIO_GROUP && req->size))
702	{
703	req->int1 = 1; /* mark request as delayed */
704	continue;
705	}
706	else
707	{
708	int res = ETP_FINISH (req);
709	if (ecb_expect_false (res))
710	return res;
711	}
712
713	if (ecb_expect_false (maxreqs && !--maxreqs))
714	break;
715
716	if (maxtime)
717	{
718	gettimeofday (&tv_now, 0);
719
720	if (tvdiff (&tv_start, &tv_now) >= maxtime)
721	break;
722	}
723	}
724
725	errno = EAGAIN;
726	return -1;
727	}
728
729	static void
730	etp_cancel (ETP_REQ *req)
731	{
732	req->cancelled = 1;
733
734	eio_grp_cancel (req);
735	}
736
737	static void
738	etp_submit (ETP_REQ *req)
739	{
740	req->pri -= ETP_PRI_MIN;
741
742	if (ecb_expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;
743	if (ecb_expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
744
745	if (ecb_expect_false (req->type == EIO_GROUP))
746	{
747	/* I hope this is worth it :/ */
748	X_LOCK (reqlock);
749	++nreqs;
750	X_UNLOCK (reqlock);
751
752	X_LOCK (reslock);
753
754	++npending;
755
756	if (!reqq_push (&res_queue, req) && want_poll_cb)
757	want_poll_cb ();
758
759	X_UNLOCK (reslock);
760	}
761	else
762	{
763	X_LOCK (reqlock);
764	++nreqs;
765	++nready;
766	reqq_push (&req_queue, req);
767	X_COND_SIGNAL (reqwait);
768	X_UNLOCK (reqlock);
769
770	etp_maybe_start_thread ();
771	}
772	}
773
774	static void ecb_cold
775	etp_set_max_poll_time (double nseconds)
776	{
777	if (WORDACCESS_UNSAFE) X_LOCK (reslock);
778	max_poll_time = nseconds * EIO_TICKS;
779	if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
780	}
781
782	static void ecb_cold
783	etp_set_max_poll_reqs (unsigned int maxreqs)
784	{
785	if (WORDACCESS_UNSAFE) X_LOCK (reslock);
786	max_poll_reqs = maxreqs;
787	if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
788	}
789
790	static void ecb_cold
791	etp_set_max_idle (unsigned int nthreads)
792	{
793	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
794	max_idle = nthreads;
795	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
796	}
797
798	static void ecb_cold
799	etp_set_idle_timeout (unsigned int seconds)
800	{
801	if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
802	idle_timeout = seconds;
803	if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
804	}
805
806	static void ecb_cold
807	etp_set_min_parallel (unsigned int nthreads)
808	{
809	if (wanted < nthreads)
810	wanted = nthreads;
811	}
812
813	static void ecb_cold
814	etp_set_max_parallel (unsigned int nthreads)
815	{
816	if (wanted > nthreads)
817	wanted = nthreads;
818
819	while (started > wanted)
820	etp_end_thread ();
821	}
822		398
823	/*****************************************************************************/	399	/*****************************************************************************/
824		400
825	static void	401	static void
826	grp_try_feed (eio_req *grp)	402	grp_try_feed (eio_req *grp)
…		…
893	}	469	}
894		470
895	void	471	void
896	eio_grp_cancel (eio_req *grp)	472	eio_grp_cancel (eio_req *grp)
897	{	473	{
898	for (grp = grp->grp_first; grp; grp = grp->grp_next)
899	eio_cancel (grp);	474	etp_grp_cancel (grp);
900	}	475	}
901		476
902	void	477	void
903	eio_cancel (eio_req *req)	478	eio_cancel (eio_req *req)
904	{	479	{
…		…
2156	strcpy (name + (len <= namelen - 4 ? len : namelen - 4), "/eio");	1731	strcpy (name + (len <= namelen - 4 ? len : namelen - 4), "/eio");
2157	prctl (PR_SET_NAME, (unsigned long)name, 0, 0, 0);	1732	prctl (PR_SET_NAME, (unsigned long)name, 0, 0, 0);
2158	#endif	1733	#endif
2159	}	1734	}
2160		1735
		1736	/* TODO: move somehow to etp.c */
2161	X_THREAD_PROC (etp_proc)	1737	X_THREAD_PROC (etp_proc)
2162	{	1738	{
2163	ETP_REQ *req;	1739	ETP_REQ *req;
2164	struct timespec ts;	1740	struct timespec ts;
2165	etp_worker self = (etp_worker )thr_arg;	1741	etp_worker self = (etp_worker )thr_arg;
…		…
2211		1787
2212	--nready;	1788	--nready;
2213		1789
2214	X_UNLOCK (reqlock);	1790	X_UNLOCK (reqlock);
2215		1791
2216	if (req->type < 0)	1792	if (req->type == ETP_TYPE_QUIT)
2217	goto quit;	1793	goto quit;
2218		1794
2219	ETP_EXECUTE (self, req);	1795	ETP_EXECUTE (self, req);
2220		1796
2221	X_LOCK (reslock);	1797	X_LOCK (reslock);

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing libeio/eio.c (file contents): Revision 1.128 by root, Mon Feb 18 03:16:30 2013 UTC vs. Revision 1.129 by root, Sun Apr 14 09:43:19 2013 UTC

Diff Legend

Comparing libeio/eio.c (file contents):
Revision 1.128 by root, Mon Feb 18 03:16:30 2013 UTC vs.
Revision 1.129 by root, Sun Apr 14 09:43:19 2013 UTC