--- /dev/null
+#/*
+# PipeBuf version 1.2
+# To compile run: sh pipebuf.c
+# Help is then available: ./pipebuf -h
+
+# Copyright (C) 1998 Jan Kratochvil <short@ucw.cz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; you must use exactly version 2.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You may download a copy of the GNU General Public License from URL
+# http://www.opensource.org/gpl-license.html
+# If not, write to the Free Software Foundation,
+# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+echo "Compiling PipeBuf..."
+a="cc"
+b="-s -Wall -O6 -fexpensive-optimizations -fomit-frame-pointer -D_GNU_SOURCE=1"
+c="-o `basename "$0" .c` $0"
+echo "$a $b $c"
+if $a $b $c;then echo -n
+else echo "$a $c"
+ if $a $c;then echo -n
+ else echo "Failed - please check the output.";exit
+ fi
+fi
+echo "done."
+exit
+*/
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/mman.h>
+#include <signal.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <string.h>
+#include <limits.h>
+#include <sys/ipc.h>
+#include <sys/shm.h>
+#include <sys/msg.h>
+#include <getopt.h>
+#include <errno.h>
+#include <time.h>
+
+#define BUFSZ (12<<10) /* in KB */
+#define BUFWARN (80) /* in percents */
+#define WARNTIM (5) /* in seconds */
+#define MAX_XFER (PIPE_BUF)
+#undef DEBUG
+
+#ifndef SHMMAX
+#define SHMMAX 0x2000000
+#endif
+
+#ifndef __NORETURN
+#if __GNUC__ >= 2
+#define __NORETURN __attribute__((__noreturn__))
+#else
+#define __NORETURN
+#endif
+#endif
+
+#define RDID (0)
+#define WRID (1)
+
+#ifndef DEBUG
+#define dbg(cmd)
+#else
+#define dbg(cmd) cmd
+#endif
+#define PNAME_LEN (16)
+#define bufw(n) ((n)==bufsz?0:(n))
+#define failf(name) do { fprintf(stderr,"%s: ",pname); perror(name"()"); exit(EXIT_FAILURE); } while (0)
+#ifndef max
+#define max(a,b) ((a)>=(b)?(a):(b))
+#endif
+#ifndef min
+#define min(a,b) ((a)<=(b)?(a):(b))
+#endif
+
+const char version[]="This is PipeBuf, version 1.2\n";
+
+long bufsz=BUFSZ<<10,bufwarn=BUFWARN;
+int prefill,quiet,verbose;
+
+char *pname;
+int shmid=-1,dis_cleanup=0,myself;
+pid_t other=-1;
+volatile struct {
+ long rp,wp;
+ int eof;
+ time_t up;
+ struct {
+ char sent;
+ int msqid;
+ } p[2];
+ } *comm;
+struct msgbuf smsgbuf={1,{0}};
+struct msgbuf rmsgbuf;
+
+static void cleanup(void)
+{
+ dbg(fprintf(stderr,"%s: cleanup()\n",pname));
+ if (dis_cleanup) return;
+ if (other!=-1) kill(other,SIGTERM);
+ shmctl(shmid,IPC_RMID,NULL);
+ if (comm) {
+ dbg(fprintf(stderr,"%s: msgctl(%d,IPC_RMID), myself=%d\n",pname,comm->p[ myself].msqid, myself));
+ msgctl(comm->p[ myself].msqid,IPC_RMID,NULL);
+ dbg(fprintf(stderr,"%s: msgctl(%d,IPC_RMID),!myself=%d\n",pname,comm->p[!myself].msqid,!myself));
+ msgctl(comm->p[!myself].msqid,IPC_RMID,NULL);
+ }
+ dis_cleanup=1;
+ exit(EXIT_FAILURE);
+}
+
+static void wake(void)
+{
+ dbg(fprintf(stderr,"%s: waking (sent=%d)...\n",pname,comm->p[myself].sent));
+ if (comm->p[myself].sent) return;
+ dbg(fprintf(stderr,"%s: msgsnd(msqid=%d), myself=%d\n",pname,comm->p[!myself].msqid,myself));
+/* dbg(fprintf(stderr,"%s: sleeping 5 sec...\n",pname));
+ sleep(5);
+ dbg(fprintf(stderr,"%s: sleeping 5 sec done\n",pname));*/
+ if (msgsnd(comm->p[!myself].msqid,(struct msgbuf *)&smsgbuf,0,IPC_NOWAIT)) if (errno!=EAGAIN) failf("msgsnd");
+ comm->p[myself].sent++;
+ dbg(fprintf(stderr,"%s: waked (sent=%d)\n",pname,comm->p[myself].sent));
+}
+
+static void shake(void)
+{
+ wake();
+ dbg(fprintf(stderr,"%s: waiting (sent=%d)...\n",pname,comm->p[myself].sent));
+ if (comm->eof) dbg(fprintf(stderr,"%s: EOF => breaking out of shake()!\n",pname));
+ else {
+ dbg(fprintf(stderr,"%s: msgrcv(msqid=%d),myself=%d\n",pname,comm->p[myself].msqid,myself));
+/* dbg(fprintf(stderr,"%s: sleeping 5 sec...\n",pname));
+ sleep(5);
+ dbg(fprintf(stderr,"%s: sleeping 5 sec done\n",pname));*/
+ if (msgrcv(comm->p[myself].msqid,&rmsgbuf,0,0,0)) failf("msgrcv");
+ }
+ comm->p[!myself].sent--;
+ dbg(if (comm->p[!myself].sent<0) {
+ fprintf(stderr,"%s: FATAL - .sent=%d (<0)!\n",pname,comm->p[!myself].sent);
+ exit(EXIT_FAILURE);
+ });
+ dbg(fprintf(stderr,"%s: wait returned (sent=%d)\n",pname,comm->p[myself].sent));
+}
+
+static void warnbuf(void)
+{
+long crp,cwp,bufused,cup;
+float percused;
+char *tim;
+ if (!comm->up||quiet||comm->eof||time(NULL)-comm->up<WARNTIM) return;
+ comm->up=cup=time(NULL);
+ crp=comm->rp; cwp=comm->wp;
+ bufused=crp-cwp+bufsz*!(crp>cwp);
+ if ((percused=(float)bufused*100/bufsz)>=bufwarn) return;
+ tim=ctime(&cup);
+ *strchr(tim,'\n')='\0';
+ fprintf(stderr,"%s: %s: WARNING - Low buffer fill-up: %8ld of %8ld (%2.1f%%)\n",
+ pname,tim,bufused,bufsz,percused);
+}
+
+static __NORETURN void usage(void)
+{
+ fprintf(stderr,"\
+%s\
+This command offers the pipe buffering:\n\
+\n\
+Usage: pipebuf [-b|--buffer <size in KB>] [-p|--prefill] [-w|--warning <percent>]\n\
+ [-q|--quiet] [-v|--verbose] [-h|--help] [-V|--version]\n\
+\n\
+ -b, --buffer <size in KB>\tSpecify buffer size (1-%dKB, def=%dKB)\n\
+ -p, --prefill\t\t\tFill the buffer before first write\n\
+ -w, --warning <percent>\tNo-buffer-data warnings threshold (0-100%%, def=%d%%)\n\
+ -q, --quiet\t\t\tDon't print warnings\n\
+ -v, --verbose\t\t\tInform about phases of transfer\n\
+ -h, --help\t\t\tPrint a summary of the options\n\
+ -V, --version\t\t\tPrint the version number\n\
+",version,(SHMMAX>>10)-1,BUFSZ,BUFWARN);
+ exit(EXIT_FAILURE);
+}
+
+const struct option longopts[]={
+{"buffer" ,1,0,'b'},
+{"prefill",0,0,'p'},
+{"warning",1,0,'w'},
+{"quiet" ,0,0,'q'},
+{"verbose",0,0,'v'},
+{"help" ,0,0,'h'},
+{"version",0,0,'V'}};
+
+int main(int argc,char **argv)
+{
+long cfp;
+int r,optc;
+caddr_t buf;
+char *s;
+
+ pname=*argv;
+ atexit(cleanup);
+ signal(SIGTERM,(void (*)(int))cleanup);
+ signal(SIGQUIT,(void (*)(int))cleanup);
+ signal(SIGINT ,(void (*)(int))cleanup);
+ signal(SIGHUP ,(void (*)(int))cleanup);
+ while ((optc=getopt_long(argc,argv,"b:pw:qvhV",longopts,NULL))!=EOF) switch (optc) {
+ case 'b':
+ errno=EINVAL;
+ bufsz=strtol(optarg,&s,0);
+ if (*s!='\0'||bufsz<1||bufsz<<10>=SHMMAX) { perror(optarg); usage(); }
+ bufsz<<=10;
+ break;
+ case 'p':
+ prefill=1;
+ break;
+ case 'w':
+ errno=EINVAL;
+ bufwarn=strtol(optarg,&s,0);
+ if (*s!='\0'||bufwarn<0||bufwarn>100) { perror(optarg); usage(); }
+ break;
+ case 'q':
+ quiet=1;
+ verbose=0;
+ break;
+ case 'v':
+ verbose=1;
+ quiet=0;
+ break;
+ case 'V':
+ fprintf(stderr,version);
+ exit(EXIT_FAILURE);
+ default: /* also 'h' */
+ usage();
+ break;
+ }
+ if ((shmid=shmget(IPC_PRIVATE,bufsz+sizeof(*comm),0600|IPC_CREAT|IPC_EXCL))==-1) failf("shmget");
+ if ((int)(buf=shmat(shmid,0,0))==-1) failf("shmat");
+ comm=(void *)buf+bufsz;
+ bzero((void *)comm,sizeof(*comm));
+ if ((comm->p[RDID].msqid=msgget(IPC_PRIVATE,0777|IPC_CREAT|IPC_EXCL))==-1) failf("msgget");
+ if ((comm->p[WRID].msqid=msgget(IPC_PRIVATE,0777|IPC_CREAT|IPC_EXCL))==-1) failf("msgget");
+ if (!prefill) comm->up=time(NULL);
+ other=fork();
+ if (other) {
+ /* Read process */
+ dbg(fprintf(stderr,"%s: started rd\n",pname));
+ myself=RDID;
+ strncat(pname,"-rd",max(PNAME_LEN-strlen(pname)-1,0));
+ if (close(STDOUT_FILENO)) failf("close");
+ if (verbose) fprintf(stderr,"%s: Using buffer %ldKB%s...\n",pname,bufsz>>10,(prefill?", filling":""));
+ dbg(fprintf(stderr,"%s: pname check rd\n",pname));
+ for (;;) {
+ if (bufw(comm->rp+1)==comm->wp) {
+ if (!comm->up) {
+ comm->up=time(NULL);
+ if (verbose) fprintf(stderr,"%s: Buffer filled-up, starting transfer...\n",pname);
+ }
+ shake();
+ continue;
+ }
+ warnbuf();
+ cfp=comm->wp;
+ dbg(fprintf(stderr,"%s: rp=%ld, wp=%ld",pname,comm->rp,cfp));
+ cfp=(cfp<=comm->rp?bufsz-comm->rp-!cfp:cfp-1-comm->rp);
+ dbg(fprintf(stderr,", read(%d,%08lx,%ld)\n",STDIN_FILENO,(long)&buf[comm->rp],min(cfp,MAX_XFER)));
+ if ((r=read(STDIN_FILENO,&buf[comm->rp],min(cfp,MAX_XFER)))==-1) failf("read");
+ dbg(fprintf(stderr,"%s: rp=%ld, wp=%ld, read=%d\n",pname,bufw(comm->rp+r),comm->wp,r));
+ if (r) comm->rp=bufw(comm->rp+r);
+ else {
+ comm->eof=1;
+ if (!comm->up) {
+ comm->up=time(NULL);
+ if (verbose) fprintf(stderr,"%s: Reached EOF before buffer fill-up, starting transfer...\n",pname);
+ }
+ wake();
+ break;
+ }
+ if (comm->up) wake();
+ }
+ if (verbose) fprintf(stderr,"%s: All input data read, waiting for write completion...\n",pname);
+ if (waitpid(other,NULL,0)!=other) failf("waitpid");
+ }
+ else {
+ /* Write process */
+ dbg(fprintf(stderr,"%s: started wr\n",pname));
+ myself=WRID;
+ strncat(pname,"-wr",max(PNAME_LEN-strlen(pname)-1,0));
+ other=getppid();
+ if (close(STDIN_FILENO)) failf("close");
+ dbg(fprintf(stderr,"%s: pname check wr\n",pname));
+ for (;;) {
+ if (comm->eof&&comm->rp==comm->wp) break;
+ while (!comm->eof&&(!comm->up||comm->rp==comm->wp)) shake();
+ if (comm->eof&&comm->rp==comm->wp) break;
+ cfp=comm->rp;
+ dbg(fprintf(stderr,"%s: rp=%ld, wp=%ld",pname,cfp,comm->wp));
+ cfp=(cfp<=comm->wp?bufsz-comm->wp:cfp-comm->wp);
+ dbg(fprintf(stderr,", write(%d,%08lx,%ld)\n",STDOUT_FILENO,(long)&buf[comm->wp],min(cfp,MAX_XFER)));
+ if ((r=write(STDOUT_FILENO,&buf[comm->wp],min(cfp,MAX_XFER)))==-1) failf("write");
+ dbg(fprintf(stderr,"%s: rp=%ld, wp=%ld, write=%d\n",pname,comm->rp,bufw(comm->wp+r),r));
+ if (!(comm->wp=bufw(comm->wp+r))&&comm->rp) continue;
+ wake();
+ warnbuf();
+ }
+ }
+ if (verbose) fprintf(stderr,"%s: Ending operation (sent=%d).\n",pname,comm->p[myself].sent);
+ return(EXIT_SUCCESS);
+}
git://git.jankratochvil.net / nethome.git / commitdiff