RPM Package Manager, CVS Repository http://rpm5.org/cvs/ ____________________________________________________________________________
Server: rpm5.org Name: Jeff Johnson Root: /v/rpm/cvs Email: j...@rpm5.org Module: rpm Date: 23-May-2017 21:36:46 Branch: rpm-5_4 Handle: 2017052319364600 Modified files: (Branch: rpm-5_4) rpm CHANGES rpm/rpmio msqio.c rpmmsq.h tmq.c Log: - msqio: permit POSIX/SysV message queue coexistence, default is POSIX. Summary: Revision Changes Path 1.3501.2.558+1 -0 rpm/CHANGES 1.1.2.8 +377 -256 rpm/rpmio/msqio.c 1.1.2.7 +29 -22 rpm/rpmio/rpmmsq.h 1.1.2.7 +18 -5 rpm/rpmio/tmq.c ____________________________________________________________________________ patch -p0 <<'@@ .' Index: rpm/CHANGES ============================================================================ $ cvs diff -u -r1.3501.2.557 -r1.3501.2.558 CHANGES --- rpm/CHANGES 22 May 2017 21:20:55 -0000 1.3501.2.557 +++ rpm/CHANGES 23 May 2017 19:36:46 -0000 1.3501.2.558 @@ -1,4 +1,5 @@ 5.4.17 -> 5.4.18: + - jbj: msqio: permit POSIX/SysV message queue coexistence, default is POSIX. - jbj: msqio: set oflags/fdno on fd. - jbj: msqio: loopback mode refactoring. - jbj: msqio: make peace with mq_notify (and sigev detached threads). @@ . patch -p0 <<'@@ .' Index: rpm/rpmio/msqio.c ============================================================================ $ cvs diff -u -r1.1.2.7 -r1.1.2.8 msqio.c --- rpm/rpmio/msqio.c 22 May 2017 21:20:56 -0000 1.1.2.7 +++ rpm/rpmio/msqio.c 23 May 2017 19:36:46 -0000 1.1.2.8 @@ -28,11 +28,11 @@ #include "debug.h" int _rpmmsq_debug; -#define F_ISSET(_f, _FLAG) (((_f) & ((RPMMSQ_FLAGS_##_FLAG) & ~0x40000000)) != RPMMSQ_FLAGS_NONE) -#define QF_ISSET(_FLAG) F_ISSET(msq->flags, _FLAG) +#define F_ISSET(_f, _FLAG) ((_f) & RPMMSQ_FLAGS_##_FLAG) +#define MSQF_ISSET(_FLAG) F_ISSET(msq->flags, _FLAG) #define SPEW(_list) \ - if (QF_ISSET(DEBUG) || _rpmmsq_debug || _rpmio_debug) fprintf _list + if (MSQF_ISSET(DEBUG) || _rpmmsq_debug || _rpmio_debug) fprintf _list #define MSQONLY(fd) assert(fdGetIo(fd) == msqio) @@ -133,116 +133,159 @@ break; } - msq->flags = flags; msq->qname = rpmGetPath("/", path, NULL); + + key_t key = 0; + if (!strcmp(msq->qname, "/private") + || !strcmp(msq->qname, "/IPC_PRIVATE")) { + key = IPC_PRIVATE; + flags |= RPMMSQ_FLAGS_DELETE; /* XXX delete on close. */ + } else + if (!strcmp(msq->qname, "/program") + || !strcmp(msq->qname, "/__progname_key")) { + key = __progname_key; + } else { + const char * lpath = msq->qname; + int projid = __progname_projid; + if (!strcmp(lpath, "/rpm")) + lpath = "/usr/lib/rpm"; + key = ftok(lpath, projid); + if (key == -1) { + int lvl = RPMLOG_WARNING; + rpmlog(lvl, "%s: ftok(%s,0x%02x) failed: %m\n", + __FUNCTION__, path, (projid & 0xff)); + key = __progname_key; + rpmlog(lvl, "Using program key 0x%x\n", key); + } + } + msq->key = key; + + msq->flags = flags; msq->qid = -1; msq->oflags = oflags; msq->omode = omode & 0777; /* XXX mask IPC_CREAT | IPC_EXCL */ msq->mtype = getpid(); /* XXX SysV message queues only. */ -#if defined(WITH_MQ) - - int xx; - /* Reset the queue by removing. */ - if (QF_ISSET(RESET)) { - xx = mq_unlink(msq->qname); -SPEW((stderr, "<-- %s(%s) rc %d\n", "mq_unlink", msq->qname, xx)); + if (MSQF_ISSET(RESET)) + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { +#if defined(WITH_MQ) + int xx = mq_unlink(msq->qname); +SPEW((stderr, "<-- %s(%s) rc %d (%s)\n", "mq_unlink", msq->qname, xx, __FUNCTION__)); +#endif /* WITH_MQ */ + } break; + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + int qid = msgget(msq->key, (omode & 0777)); + if (qid != -1) { + int cmd = IPC_RMID; + int xx = msgctl(qid, cmd, NULL); +SPEW((stderr, "<-- %s(0x%x,0%o,%p) rc %d (%s)\n", "msqctl", qid, cmd, NULL, xx, __FUNCTION__)); + } +#endif /* WITH_MSQ */ + } break; + default: + break; } - struct mq_attr _attrs = { - .mq_flags = (oflags & O_NONBLOCK), - .mq_maxmsg = 10, /* /proc/sys/fs/mqueue/msg_default */ - .mq_msgsize = 8192, /* /proc/sys/fs/mqueue/msgsize_default */ - .mq_curmsgs = 0, - }, *attrs = &_attrs; - msq->qid = mq_open(msq->qname, msq->oflags, msq->omode, attrs); + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { +#if defined(WITH_MQ) + struct mq_attr _attrs = { + .mq_flags = (oflags & O_NONBLOCK), + .mq_maxmsg = 10, /* /proc/sys/fs/mqueue/msg_default */ + .mq_msgsize = 8192, /* /proc/sys/fs/mqueue/msgsize_default */ + .mq_curmsgs = 0, + }, *attrs = &_attrs; + msq->qid = mq_open(msq->qname, msq->oflags, msq->omode, attrs); SPEW((stderr, "<-- %s(%s,0x%x,0%o,%p) qid %d\n", "mq_open", msq->qname, msq->oflags, msq->omode, attrs, msq->qid)); - /* (loopback mode) Configure the detached reader. */ - if (msq->qid != -1 && QF_ISSET(LOOP)) { - pthread_attr_t attr; - xx = pthread_getattr_default_np(&attr); - xx = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + /* (loopback mode) Configure the detached reader. */ + if (msq->qid != -1 && MSQF_ISSET(LOOP)) { + pthread_attr_t attr; + int xx; + xx = pthread_getattr_default_np(&attr); + xx = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); - struct sigevent sigev = { + struct sigevent sigev = { .sigev_notify = SIGEV_THREAD, .sigev_notify_function = rpmmsqReader, .sigev_notify_attributes = &attr, .sigev_value.sival_ptr = msq, - }; - xx = rpmmsqNotify(msq, &sigev); - } - -#elif defined(WITH_MSQ) - - if (!strcmp(msq->qname, "/private") - || !strcmp(msq->qname, "/IPC_PRIVATE")) - msq->key = IPC_PRIVATE; - else - if (!strcmp(msq->qname, "/program") - || !strcmp(msq->qname, "/__progname_key")) - msq->key = __progname_key; - else { - const char * lpath = msq->qname; - int projid = __progname_projid; - if (!strcmp(lpath, "/rpm")) - lpath = "/usr/lib/rpm"; - msq->key = ftok(lpath, projid); - if (msq->key == -1) { - int lvl = RPMLOG_WARNING; - rpmlog(lvl, "%s: ftok(%s,0x%02x) failed: %m\n", - __FUNCTION__, path, (projid & 0xff)); - msq->key = __progname_key; - rpmlog(lvl, "Using program key 0x%x\n", msq->key); + }; + xx = rpmmsqNotify(msq, &sigev); } - } - msq->qid = msgget(msq->key, omode); +#endif /* WITH_MQ */ + } break; + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + msq->qid = msgget(msq->key, omode); SPEW((stderr, "<-- %s(0x%x,0%o) qid %d\n", "msqget", msq->key, omode, msq->qid)); +#endif /* WITH_MSQ */ + } break; + default: + break; + } -#endif /* WITH_MQ */ +SPEW((stderr, "<-- %s(%s,%s,%d,0x%x) qid %d\n", __FUNCTION__, path, fmode, fdno, flags, (msq ? msq->qid : -1))); - return (msq->qid != -1 ? rpmmsqLink(msq) : rpmmsqFree(msq)); + return rpmmsqLink(msq); } ssize_t rpmmsqRecv(rpmmsq msq, char * buf, size_t count, unsigned short *priop) { ssize_t rc = -1; /* assume failure */ + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { #if defined(WITH_MQ) + if (MSQF_ISSET(LOOP)) return 0; /* XXX EOF with loopback service. */ - if (QF_ISSET(LOOP)) return 0; /* XXX EOF with loopback service. */ - - unsigned int prio = 0; - int nb = 0; - rc = mq_receive(msq->qid, buf, count, &prio); - if (rc >= 0) { - nb = rc; - msq->nrecv++; - } + unsigned int prio = 0; + int nb = 0; + rc = mq_receive(msq->qid, buf, count, &prio); + if (rc >= 0) { + nb = rc; + msq->nrecv++; + } SPEW((stderr, "<-- %s(0x%x,%p[%lu],%p) rc %ld prio %u\t\"%.*s\"\n", "mq_receive", msq->qid, buf, (unsigned long)count, priop, (long)rc, prio, nb, buf)); - if (priop) - *priop = prio; - -#elif defined(WITH_MSQ) - - struct msgbuf * msgp = xmalloc(sizeof(*msgp) + count); - size_t msgsz = count; - long msgtyp = msq->mtype; - int msgflg = 0; - int nb = 0; + if (priop) + *priop = prio; +#endif /* WITH_MQ */ + } break; - rc = msgrcv(msq->qid, msgp, msgsz, msgtyp, msgflg); - if (rc >= 0) { - if (rc > 0) - memcpy(buf, msgp->mtext, rc); - nb = rc; - msq->nrecv++; - } + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + struct msgbuf * msgp = xmalloc(sizeof(*msgp) + count); + size_t msgsz = count; + long msgtyp = msq->mtype; + int msgflg = 0; + int nb = 0; + + rc = msgrcv(msq->qid, msgp, msgsz, msgtyp, msgflg); + if (rc >= 0) { + if (rc > 0) + memcpy(buf, msgp->mtext, rc); + nb = rc; + msq->nrecv++; + } SPEW((stderr, "<-- %s(0x%x,%p,%lu,%ld,%d) rc %ld\t\"%.*s\"\n", "msqrcv", msq->qid, msgp, (unsigned long)msgsz, (long)msgtyp, msgflg, (long)rc, nb, buf)); - msgp = _free(msgp); - -#endif /* WITH_MQ */ + msgp = _free(msgp); +#endif /* WITH_MSQ */ + } break; + default: + break; + } return rc; } @@ -251,47 +294,54 @@ { ssize_t rc = -1; /* assume failure */ + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { #if defined(WITH_MQ) - - rc = mq_send(msq->qid, buf, count, prio); - int nb = (rc == 0 ? count : 0); + rc = mq_send(msq->qid, buf, count, prio); + int nb = (rc == 0 ? count : 0); SPEW((stderr, "<-- %s(0x%x,%p[%lu],%u) rc %ld\t\"%.*s\"\n", "mq_send", msq->qid, buf, (unsigned long)count, prio, (long)rc, nb, buf)); - if (rc == 0) { /* XXX remap to write(2) return */ - rc = count; - msq->nsent++; - } - - /* (loopback mode) Wait for rpmmsqReader to start up. */ - sched_yield(); /* Give detached threads a chance to run. */ - if (QF_ISSET(LOOP) && prio == 0 && msq->tid == 0) { - struct timespec req = { 0, 1000*1000 }; - while (msq->tid == 0) { - nanosleep(&req, NULL); - msq->nwait++; + if (rc == 0) { /* XXX remap to write(2) return */ + rc = count; + msq->nsent++; } - } -#elif defined(WITH_MSQ) - - struct msgbuf * msgp = xmalloc(sizeof(*msgp) + count); - size_t msgsz = count; - int msgflg = 0; - int nb = 0; - - msgp->mtype = msq->mtype; - if (count) - memcpy(msgp->mtext, buf, count); - rc = msgsnd(msq->qid, msgp, msgsz, msgflg); - int nb = (rc == 0 ? count : 0); + /* (loopback mode) Wait for rpmmsqReader to start up. */ + sched_yield(); /* Give detached threads a chance to run. */ + if (MSQF_ISSET(LOOP) && prio == 0 && msq->tid == 0) { + struct timespec req = { 0, 1000*1000 }; + while (msq->tid == 0) { + nanosleep(&req, NULL); + msq->nwait++; + } + } +#endif /* WITH_MQ */ + } break; + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + struct msgbuf * msgp = xmalloc(sizeof(*msgp) + count); + size_t msgsz = count; + int msgflg = 0; + + msgp->mtype = msq->mtype; + if (count) + memcpy(msgp->mtext, buf, count); + rc = msgsnd(msq->qid, msgp, msgsz, msgflg); + int nb = (rc == 0 ? count : 0); SPEW((stderr, "<-- %s(0x%x,%p,%lu,%d) rc %ld\t\"%.*s\"\n", "msqsnd", msq->qid, msgp, (unsigned long)msgsz, msgflg, (long)rc, nb, buf)); - msgp = _free(msgp); - if (rc == 0) { /* XXX remap to write(2) return */ - rc = count; - msq->nsent++; + msgp = _free(msgp); + if (rc == 0) { /* XXX remap to write(2) return */ + rc = count; + msq->nsent++; + } +#endif /* WITH_MSQ */ + } break; + default: + break; } -#endif /* WITH_MQ */ - return rc; } @@ -306,43 +356,55 @@ { int rc = -2; /* assume failure */ + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { #if defined(WITH_MQ) - - if (QF_ISSET(LOOP)) { - /* (loopback mode) Terminate rpmmsqReader. */ - if (msq->tid) { - rc = rpmmsqSend(msq, __FUNCTION__, sizeof(__FUNCTION__)-1, RPMMSQ_PRIO_EXIT); - struct timespec req = { 0, 1000*1000 }; - do { - nanosleep(&req, NULL); - msq->nwait++; - } while (msq->tid != 0); + if (MSQF_ISSET(LOOP)) { + /* (loopback mode) Terminate rpmmsqReader. */ + if (msq->tid) { + rc = rpmmsqSend(msq, __FUNCTION__, sizeof(__FUNCTION__)-1, RPMMSQ_PRIO_EXIT); + struct timespec req = { 0, 1000*1000 }; + do { + nanosleep(&req, NULL); + msq->nwait++; + } while (msq->tid != 0); + } + /* (loopback mode) Turn off the sigev detached thread. */ + rc = rpmmsqNotify(msq, NULL); } - /* (loopback mode) Turn off the sigev detached thread. */ - rc = rpmmsqNotify(msq, NULL); - } SPEW((stderr, "%s: sent %d recv %d wait %d timeout %d\n", __FUNCTION__, msq->nsent, msq->nrecv, msq->nwait, msq->ntimeout)); - if (_rpmmsq_debug) - rpmmsqDump(__FUNCTION__, msq, NULL); - rc = mq_close(msq->qid); + if (_rpmmsq_debug) + rpmmsqDump(__FUNCTION__, msq, NULL); + rc = mq_close(msq->qid); SPEW((stderr, "<-- %s(0x%x) rc %d\n", "mq_close", msq->qid, rc)); - if (!rc && delete && msq->qname && !strcmp(msq->qname, "/private")) { - rc = mq_unlink(msq->qname); /* XXX rpmmsqReset? */ -SPEW((stderr, "<-- %s(%s) rc %d\n", "mq_unlink", msq->qname, rc)); - } - rc = 0; /* XXX */ - -#elif defined(WITH_MSQ) - - if (_rpmmsq_debug) - rpmmsqDump(__FUNCTION__, msq, NULL); - if (delete || msq->key == IPC_PRIVATE) - rc = rpmmsqCtl(msq, IPC_RMID, NULL); /* XXX rpmmsqReset? */ - rc = 0; /* XXX */ - + if (!rc && (delete || MSQF_ISSET(DELETE))) { + rc = mq_unlink(msq->qname); /* XXX rpmmsqReset? */ +SPEW((stderr, "<-- %s(%s) rc %d (%s)\n", "mq_unlink", msq->qname, rc, __FUNCTION__)); + } + rc = 0; /* XXX */ #endif /* WITH_MQ */ + } break; + + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + if (_rpmmsq_debug) + rpmmsqDump(__FUNCTION__, msq, NULL); + if (delete || MSQF_ISSET(DELETE)) { + int cmd = IPC_RMID; + rc = msgctl(msq->qid, cmd, NULL); +SPEW((stderr, "<-- %s(0x%x,0%o,%p) rc %d (%s)\n", "msqctl", msq->qid, cmd, NULL, rc, __FUNCTION__)); + } + rc = 0; /* XXX */ +#endif /* WITH_MSQ */ + } break; + default: + break; + } return rc; } @@ -366,46 +428,48 @@ int rpmmsqCtl(void * _msq, int cmd, void *buf) { + rpmmsq msq = (rpmmsq) _msq; int rc = -2; /* assume failure */ -#if defined(WITH_MQ) -#elif defined(WITH_MSQ) - rpmmsq msq = (rpmmsq) _msq; - if (msq) - switch (cmd) { - default: - break; - case IPC_STAT: - rc = msgctl(msq->qid, cmd, buf); - break; - case IPC_SET: - rc = msgctl(msq->qid, cmd, buf); - break; - case IPC_RMID: - rc = msgctl(msq->qid, cmd, NULL); - break; + if (msq) /* XXX WTF? */ + switch (msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { + } break; + + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + if (msq) + switch (cmd) { + default: + break; #if defined(IPC_INFO) - case IPC_INFO: - rc = msgctl(msq->qid, cmd, buf); - /* XXX rc is index of highest used message queue. */ - break; + case IPC_INFO: + /* XXX rc is index of highest used message queue. */ #endif #if defined(MSG_INFO) - case MSG_INFO: - rc = msgctl(msq->qid, cmd, buf); - /* XXX rc is index of highest used message queue. */ - break; + case MSG_INFO: + /* XXX rc is index of highest used message queue. */ #endif #if defined(MSG_STAT) - case MSG_STAT: - /* XXX 1st arg is kernel index, not qid. */ - rc = msgctl(msq->qid, cmd, buf); - /* XXX rc is the qid. */ - break; + case MSG_STAT: + /* XXX 1st arg is kernel index, not qid. */ + /* XXX rc is the qid. */ #endif + case IPC_STAT: + case IPC_SET: + case IPC_RMID: + rc = msgctl(msq->qid, cmd, buf); + break; + } +SPEW((stderr, "<-- %s(0x%x,%d,%p) rc %d\n", "msqctl", (unsigned)msq->qid, cmd, buf, rc)); +#endif /* WITH_MSQ */ + } break; + default: + break; } -SPEW((stderr, "<-- %s(0x%x,%d,%p) rc %d\n", "msqctl", (msq ? (unsigned)msq->qid : 0xdeadbeef), cmd, buf, rc)); -#endif /* WITH_MQ */ return rc; } @@ -414,13 +478,23 @@ { int rc = -2; /* assume failure */ + if (msq) /* XXX WTF? */ + switch(msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { #if defined(WITH_MQ) - if (msq) { /* XXX WTF? */ rc = mq_notify(msq->qid, _sevp); SPEW((stderr, "<-- %s(0x%x,%p) rc %d\n", "mq_notify", msq->qid, _sevp, rc)); - } -#elif defined(WITH_MSQ) #endif /* WITH_MQ */ + } break; + + case RPMMSQ_TYPE_SYSV: + { + } break; + default: + break; + } return rc; } @@ -464,94 +538,141 @@ #endif /* WITH_MQ */ } -int rpmmsqDump(const char * msg, void *_msq, FILE *fp) +static void rpmmsqDumpST(const char * msg, void * _ptr, FILE * fp) { - rpmmsq msq = (rpmmsq) _msq; - int rc = -2; /* assume failure */ + struct stat *st = (struct stat *) _ptr; - if (fp == NULL) - fp = stderr; + if (fp == NULL) fp = stderr; + if (msg) fprintf(fp, "================ %s\n", msg); - if (msg) - fprintf(fp, "================ %s\n", msg); - -#if defined(WITH_MQ) -#if defined(__linux__) - const char * lpath = rpmGetPath("/dev/mqueue/", msq->qname, NULL); - struct stat st; - rc = Stat(lpath, &st); - if (!rc) { - fprintf(fp, "\t dev: 0x%lx\n", (unsigned long)st.st_dev); - fprintf(fp, "\t ino: %lu\n", (unsigned long)st.st_ino); - fprintf(fp, "\t mode: 0%o\n", st.st_mode); - fprintf(fp, "\t nlink: %lu\n", (unsigned long)st.st_nlink); - fprintf(fp, "\t uid: %d\n", st.st_uid); - fprintf(fp, "\t gid: %d\n", st.st_gid); - fprintf(fp, "\t rdev: 0x%lx\n", (unsigned long)st.st_rdev); - fprintf(fp, "\t size: %lu\n", (unsigned long)st.st_size); - fprintf(fp, "\tblksize: %lu\n", (unsigned long)st.st_blksize); - fprintf(fp, "\t blocks: %lu\n", (unsigned long)st.st_blocks); - fprintf(fp, "\t atime: %lu\n", (unsigned long)st.st_atime); - fprintf(fp, "\t mtime: %lu\n", (unsigned long)st.st_mtime); - fprintf(fp, "\t ctime: %lu\n", (unsigned long)st.st_mtime); - } - /* XXX .fdio avoids select/poll issues on /dev/mqueue with .ufdio. */ - { FD_t fd = Fopen(lpath, "r.fdio"); - if (fd) { - unsigned char b[BUFSIZ]; - size_t nb = sizeof(b); - size_t nr = Fread(b, 1, nb-1, fd); - b[nr] = '\0'; - fprintf(stderr, "\t%.*s", (int)nr, b); - (void) Fclose(fd); - } + if (st) { + fprintf(fp, "\t dev: 0x%lx\n", (unsigned long)st->st_dev); + fprintf(fp, "\t ino: %lu\n", (unsigned long)st->st_ino); + fprintf(fp, "\t mode: 0%o\n", st->st_mode); + fprintf(fp, "\t nlink: %lu\n", (unsigned long)st->st_nlink); + fprintf(fp, "\t uid: %d\n", st->st_uid); + fprintf(fp, "\t gid: %d\n", st->st_gid); + fprintf(fp, "\t rdev: 0x%lx\n", (unsigned long)st->st_rdev); + fprintf(fp, "\t size: %lu\n", (unsigned long)st->st_size); + fprintf(fp, "\tblksize: %lu\n", (unsigned long)st->st_blksize); + fprintf(fp, "\t blocks: %lu\n", (unsigned long)st->st_blocks); + fprintf(fp, "\t atime: %lu\n", (unsigned long)st->st_atime); + fprintf(fp, "\t mtime: %lu\n", (unsigned long)st->st_mtime); + fprintf(fp, "\t ctime: %lu\n", (unsigned long)st->st_mtime); } - lpath = _free(lpath); -#endif /* __linux__ */ -#elif defined(WITH_MSQ) - struct msqid_ds ds; - rc = rpmmsqCtl(msq, IPC_STAT, &ds); - if (!rc) { +} + +static void rpmmsqDumpDS(const char * msg, void * _ptr, FILE * fp) +{ + if (fp == NULL) fp = stderr; + if (msg) fprintf(fp, "================ %s\n", msg); + +#if defined(WITH_MSQ) + struct msqid_ds *ds = (struct msqid_ds *) _ptr;; + if (ds) { #if defined(__linux__) - fprintf(fp, "\t key: 0x%08x\n", ds.msg_perm.__key); -#else - fprintf(fp, "\t key: 0x%08x\n", msq->key); + fprintf(fp, "\t key: 0x%08x\n", ds->msg_perm.__key); #endif - fprintf(fp, "\t uid: %d\n", ds.msg_perm.uid); - fprintf(fp, "\t gid: %d\n", ds.msg_perm.gid); - fprintf(fp, "\t cuid: %d\n", ds.msg_perm.cuid); - fprintf(fp, "\t cgid: %d\n", ds.msg_perm.cgid); - fprintf(fp, "\t mode: 0%o\n", ds.msg_perm.mode); + fprintf(fp, "\t uid: %d\n", ds->msg_perm.uid); + fprintf(fp, "\t gid: %d\n", ds->msg_perm.gid); + fprintf(fp, "\t cuid: %d\n", ds->msg_perm.cuid); + fprintf(fp, "\t cgid: %d\n", ds->msg_perm.cgid); + fprintf(fp, "\t mode: 0%o\n", ds->msg_perm.mode); #if defined(__linux__) - fprintf(fp, "\t seq: %d\n", ds.msg_perm.__seq); + fprintf(fp, "\t seq: %d\n", ds->msg_perm.__seq); #endif - fprintf(fp, "\t stime: %lu\n", ds.msg_stime); - fprintf(fp, "\t rtime: %lu\n", ds.msg_rtime); - fprintf(fp, "\t ctime: %lu\n", ds.msg_ctime); + fprintf(fp, "\t stime: %lu\n", ds->msg_stime); + fprintf(fp, "\t rtime: %lu\n", ds->msg_rtime); + fprintf(fp, "\t ctime: %lu\n", ds->msg_ctime); #if defined(__linux__) - fprintf(fp, "\t cbytes: %lu\n", ds.__msg_cbytes); + fprintf(fp, "\t cbytes: %lu\n", ds->__msg_cbytes); #endif - fprintf(fp, "\t qnum: %lu\n", ds.msg_qnum); - fprintf(fp, "\t qbytes: %lu\n", ds.msg_qbytes); - fprintf(fp, "\t lspid: %d\n", ds.msg_lspid); - fprintf(fp, "\t lrpid: %d\n", ds.msg_lrpid); - } - -#if defined(IPC_INFO) /* XXX linux-only */ - struct msginfo mi; - rc = rpmmsqCtl(msq, IPC_INFO, (struct msqid_ds *)&mi); - if (!rc) { - fprintf(fp, "\tmsgpool: %7d Kb\n", mi.msgpool); - fprintf(fp, "\t msgmap: %7d\n", mi.msgmap); - fprintf(fp, "\t msgmax: %7d\n", mi.msgmax); - fprintf(fp, "\t msgmnb: %7d b\n", mi.msgmnb); - fprintf(fp, "\t msgmni: %7d\n", mi.msgmni); - fprintf(fp, "\t msgssz: %7d\n", mi.msgssz); - fprintf(fp, "\t msgtql: %7d\n", mi.msgtql); - fprintf(fp, "\t msgseg: %7d\n", mi.msgseg); + fprintf(fp, "\t qnum: %lu\n", ds->msg_qnum); + fprintf(fp, "\t qbytes: %lu\n", ds->msg_qbytes); + fprintf(fp, "\t lspid: %d\n", ds->msg_lspid); + fprintf(fp, "\t lrpid: %d\n", ds->msg_lrpid); + } +#endif /* WITH_MSQ */ +} + +static void rpmmsqDumpMI(const char * msg, void * _ptr, FILE * fp) +{ + if (fp == NULL) fp = stderr; + if (msg) fprintf(fp, "================ %s\n", msg); + +#if defined(WITH_MSQ) && defined(IPC_INFO) + struct msginfo * mi = (struct msginfo *) _ptr;; + if (mi) { + fprintf(fp, "\tmsgpool: %7d Kb\n", mi->msgpool); + fprintf(fp, "\t msgmap: %7d\n", mi->msgmap); + fprintf(fp, "\t msgmax: %7d\n", mi->msgmax); + fprintf(fp, "\t msgmnb: %7d b\n", mi->msgmnb); + fprintf(fp, "\t msgmni: %7d\n", mi->msgmni); + fprintf(fp, "\t msgssz: %7d\n", mi->msgssz); + fprintf(fp, "\t msgtql: %7d\n", mi->msgtql); + fprintf(fp, "\t msgseg: %7d\n", mi->msgseg); } -#endif +#endif /* WITH_MSQ */ +} + +int rpmmsqDump(const char * msg, void *_msq, FILE *fp) +{ + rpmmsq msq = (rpmmsq) _msq; + int rc = -2; /* assume failure */ + + if (fp == NULL) fp = stderr; + if (msg) fprintf(fp, "================ %s\n", msg); + + if (msq) /* XXX WTF? */ + switch(msq->flags & RPMMSQ_TYPE_MASK) { + case RPMMSQ_TYPE_DEFAULT: + case RPMMSQ_TYPE_POSIX: + { +#if defined(WITH_MQ) +#if defined(__linux__) + const char * lpath = rpmGetPath("/dev/mqueue/", msq->qname, NULL); + struct stat sb; + + rc = Stat(lpath, &sb); + if (!rc) + rpmmsqDumpST(NULL, &sb, NULL); + + /* XXX .fdio avoids select/poll issues on /dev/mqueue with .ufdio. */ + { FD_t fd = Fopen(lpath, "r.fdio"); + if (fd) { + unsigned char b[BUFSIZ]; + size_t nb = sizeof(b); + size_t nr = Fread(b, 1, nb-1, fd); + b[nr] = '\0'; + fprintf(stderr, "\t%.*s", (int)nr, b); + (void) Fclose(fd); + } + } + + lpath = _free(lpath); +#endif /* __linux__ */ #endif /* WITH_MQ */ + } break; + + case RPMMSQ_TYPE_SYSV: + { +#if defined(WITH_MSQ) + struct msqid_ds ds; + rc = rpmmsqCtl(msq, IPC_STAT, &ds); + if (!rc) + rpmmsqDumpDS(NULL, &ds, NULL); + +#if defined(IPC_INFO) /* XXX linux-only? */ + struct msginfo mi; + rc = rpmmsqCtl(msq, IPC_INFO, (struct msqid_ds *)&mi); + if (!rc) + rpmmsqDumpMI(NULL, &mi, NULL); +#endif +#endif /* WITH_MSQ */ + } break; + default: + break; + } return rc; } @@ . patch -p0 <<'@@ .' Index: rpm/rpmio/rpmmsq.h ============================================================================ $ cvs diff -u -r1.1.2.6 -r1.1.2.7 rpmmsq.h --- rpm/rpmio/rpmmsq.h 22 May 2017 21:07:19 -0000 1.1.2.6 +++ rpm/rpmio/rpmmsq.h 23 May 2017 19:36:46 -0000 1.1.2.7 @@ -9,16 +9,28 @@ /** */ +typedef enum rpmmsqType_e { + RPMMSQ_TYPE_DEFAULT = 0, + RPMMSQ_TYPE_POSIX = 1, + RPMMSQ_TYPE_SYSV = 2, +} rpmmsqType; +#define RPMMSQ_TYPE_MASK 0x3 + +/** + */ #define _KFB(n) (1U << (n)) -#define _MFB(n) (_KFB(n) | 0x40000000) +#define _MFB(n) (_KFB(n)) typedef enum rpmmsqFlags_e { RPMMSQ_FLAGS_NONE = 0, - - RPMMSQ_FLAGS_INFO = _MFB( 0), - RPMMSQ_FLAGS_LOOP = _MFB( 1), - RPMMSQ_FLAGS_RESET = _MFB( 2), - RPMMSQ_FLAGS_DEBUG = _MFB( 3), - + RPMMSQ_FLAGS_POSIX = RPMMSQ_TYPE_POSIX, + RPMMSQ_FLAGS_SYSV = RPMMSQ_TYPE_SYSV, + /* bits 2-7 reserved */ + RPMMSQ_FLAGS_INFO = _MFB( 8), + RPMMSQ_FLAGS_LOOP = _MFB( 9), + RPMMSQ_FLAGS_RESET = _MFB(10), + RPMMSQ_FLAGS_DELETE = _MFB(11), + RPMMSQ_FLAGS_DEBUG = _MFB(12), + /* bits 13-31 unused */ } rpmmsqFlags; #undef _MFB #undef _KFB @@ -28,23 +40,18 @@ #if defined(_RPMMSQ_INTERNAL) struct rpmmsq_s { struct rpmioItem_s _item; /*!< usage mutex and pool identifier. */ - rpmmsqFlags flags; - + rpmmsqFlags flags; /*!< message queue option bits. */ const char * qname; /*!> message queue path. */ int qid; /*!< message queue id. */ - - int oflags; - int omode; - - key_t key; /*!< msgget arg */ - long mtype; - - volatile pthread_t tid; /*!< sigev thread id. */ - int ntimeout; - int nwait; - - int nsent; - int nrecv; + int oflags; /*!< message queue open flags. */ + int omode; /*!< message queue open mode. */ + key_t key; /*!< SysV: ftok(3) or IPC_PRIVATE. */ + long mtype; /*!< SysV: message type. */ + volatile pthread_t tid; /*!< LOOP: sigev thread id. */ + int ntimeout; /*!< LOOP: no. of receive timeouts. */ + int nwait; /*!< LOOP: no. of waits for thread change. */ + int nsent; /*!< no. messages sent. */ + int nrecv; /*!< no. messages received. */ }; #endif /* _RPMMSQ_INTERNAL */ @@ . patch -p0 <<'@@ .' Index: rpm/rpmio/tmq.c ============================================================================ $ cvs diff -u -r1.1.2.6 -r1.1.2.7 tmq.c --- rpm/rpmio/tmq.c 22 May 2017 21:07:19 -0000 1.1.2.6 +++ rpm/rpmio/tmq.c 23 May 2017 19:36:46 -0000 1.1.2.7 @@ -50,9 +50,12 @@ static rpmmsqFlags flags = #ifdef DYING RPMMSQ_FLAGS_DEBUG | + RPMMSQ_FLAGS_DELETE | RPMMSQ_FLAGS_RESET | RPMMSQ_FLAGS_LOOP | RPMMSQ_FLAGS_INFO | + RPMMSQ_FLAGS_SYSV | + RPMMSQ_FLAGS_POSIX | #endif 0; @@ -382,7 +385,7 @@ xx = rpmmsqRecv(msq, b, nb, NULL); } - xx = rpmmsqClose(msq, 1); + xx = rpmmsqClose(msq, 0); msq = rpmmsqFree(msq); /* XXX mq_notify memory leak if loopback mode. */ @@ -407,17 +410,26 @@ #define POPT_BIT_XOR (POPT_ARG_VAL|POPT_ARGFLAG_XOR) #endif +#define _MSQBIT(_v) (RPMMSQ_FLAGS_##_v) static struct poptOption rpmmqOptionsTable[] = { { "debug", 'd', POPT_ARG_VAL|POPT_ARGFLAG_DOC_HIDDEN, &_debug, 1, NULL, NULL }, - { "info", 'i', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, RPMMSQ_FLAGS_INFO, + + { "delete", 'D', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, _MSQBIT(DELETE), + N_("remove queue after closing"), NULL }, + { "info", 'i', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, _MSQBIT(INFO), N_("display queue info on close"), NULL }, - { "loop", 'l', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, RPMMSQ_FLAGS_LOOP, + { "loop", 'l', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, _MSQBIT(LOOP), N_("toggle mq_notify running"), NULL }, { "priority", 'p', POPT_ARG_INT, &priority, 0, N_("send message <priority>"), N_("<priority>") }, - { "reset", 'r', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, RPMMSQ_FLAGS_RESET, - N_("remove and recreate queue when opening"), NULL }, + { "reset", 'R', POPT_BIT_XOR|POPT_ARGFLAG_TOGGLE, &flags, _MSQBIT(RESET), + N_("remove queue before opening"), NULL }, + + { "posix", 'P', POPT_BIT_SET, &flags, _MSQBIT(POSIX), + N_("use POSIX message queues"), NULL }, + { "sysv", 'S', POPT_BIT_SET, &flags, _MSQBIT(SYSV), + N_("use SysV message queues"), NULL }, { NULL, '\0', POPT_ARG_INCLUDE_TABLE, rpmioAllPoptTable, 0, N_(" Common options for all rpmio executables:"), NULL }, @@ -426,6 +438,7 @@ POPT_AUTOHELP POPT_TABLEEND }; +#undef _MSQBIT int main(int argc, char *argv[]) @@ . ______________________________________________________________________ RPM Package Manager http://rpm5.org CVS Sources Repository rpm-cvs@rpm5.org