Lines Matching defs:mq
258 static void mqueue_free(struct mqueue *mq);
259 static int mqueue_send(struct mqueue *mq, const char *msg_ptr,
262 static int mqueue_receive(struct mqueue *mq, char *msg_ptr,
265 static int _mqueue_send(struct mqueue *mq, struct mqueue_msg *msg,
267 static int _mqueue_recv(struct mqueue *mq, struct mqueue_msg **msg,
269 static void mqueue_send_notification(struct mqueue *mq);
991 struct mqueue *mq;
997 mq = mqueue_alloc(NULL);
998 if (mq == NULL)
1014 pn->mn_data = mq;
1017 mqueue_free(mq);
1344 struct mqueue *mq;
1350 mq = VTOMQ(vp);
1353 mq->mq_totalbytes,
1354 mq->mq_maxmsg,
1355 mq->mq_curmsgs,
1356 mq->mq_msgsize);
1593 struct mqueue *mq;
1597 mq = uma_zalloc(mqueue_zone, M_WAITOK | M_ZERO);
1598 TAILQ_INIT(&mq->mq_msgq);
1600 mq->mq_maxmsg = attr->mq_maxmsg;
1601 mq->mq_msgsize = attr->mq_msgsize;
1603 mq->mq_maxmsg = default_maxmsg;
1604 mq->mq_msgsize = default_msgsize;
1606 mtx_init(&mq->mq_mutex, "mqueue lock", NULL, MTX_DEF);
1607 knlist_init_mtx(&mq->mq_rsel.si_note, &mq->mq_mutex);
1608 knlist_init_mtx(&mq->mq_wsel.si_note, &mq->mq_mutex);
1610 return (mq);
1617 mqueue_free(struct mqueue *mq)
1621 while ((msg = TAILQ_FIRST(&mq->mq_msgq)) != NULL) {
1622 TAILQ_REMOVE(&mq->mq_msgq, msg, msg_link);
1626 mtx_destroy(&mq->mq_mutex);
1627 seldrain(&mq->mq_rsel);
1628 seldrain(&mq->mq_wsel);
1629 knlist_destroy(&mq->mq_rsel.si_note);
1630 knlist_destroy(&mq->mq_wsel.si_note);
1631 uma_zfree(mqueue_zone, mq);
1689 mqueue_send(struct mqueue *mq, const char *msg_ptr,
1700 if (msg_len > mq->mq_msgsize)
1708 error = _mqueue_send(mq, msg, -1);
1716 error = _mqueue_send(mq, msg, 0);
1723 error = _mqueue_send(mq, msg, -1);
1742 error = _mqueue_send(mq, msg, tvtohz(&tv));
1757 _mqueue_send(struct mqueue *mq, struct mqueue_msg *msg, int timo)
1762 mtx_lock(&mq->mq_mutex);
1763 while (mq->mq_curmsgs >= mq->mq_maxmsg && error == 0) {
1765 mtx_unlock(&mq->mq_mutex);
1768 mq->mq_senders++;
1769 error = msleep(&mq->mq_senders, &mq->mq_mutex,
1771 mq->mq_senders--;
1775 if (mq->mq_curmsgs >= mq->mq_maxmsg) {
1776 mtx_unlock(&mq->mq_mutex);
1780 if (TAILQ_EMPTY(&mq->mq_msgq)) {
1781 TAILQ_INSERT_HEAD(&mq->mq_msgq, msg, msg_link);
1783 if (msg->msg_prio <= TAILQ_LAST(&mq->mq_msgq, msgq)->msg_prio) {
1784 TAILQ_INSERT_TAIL(&mq->mq_msgq, msg, msg_link);
1786 TAILQ_FOREACH(msg2, &mq->mq_msgq, msg_link) {
1793 mq->mq_curmsgs++;
1794 mq->mq_totalbytes += msg->msg_size;
1795 if (mq->mq_receivers)
1796 wakeup_one(&mq->mq_receivers);
1797 else if (mq->mq_notifier != NULL)
1798 mqueue_send_notification(mq);
1799 if (mq->mq_flags & MQ_RSEL) {
1800 mq->mq_flags &= ~MQ_RSEL;
1801 selwakeup(&mq->mq_rsel);
1803 KNOTE_LOCKED(&mq->mq_rsel.si_note, 0);
1804 mtx_unlock(&mq->mq_mutex);
1813 mqueue_send_notification(struct mqueue *mq)
1820 mtx_assert(&mq->mq_mutex, MA_OWNED);
1821 nt = mq->mq_notifier;
1826 mq->mq_notifier = NULL;
1835 mq->mq_notifier = NULL;
1844 mqueue_receive(struct mqueue *mq, char *msg_ptr,
1853 if (msg_len < mq->mq_msgsize)
1858 error = _mqueue_recv(mq, &msg, -1);
1866 error = _mqueue_recv(mq, &msg, 0);
1873 error = _mqueue_recv(mq, &msg, -1);
1893 error = _mqueue_recv(mq, &msg, tvtohz(&tv));
1914 _mqueue_recv(struct mqueue *mq, struct mqueue_msg **msg, int timo)
1918 mtx_lock(&mq->mq_mutex);
1919 while ((*msg = TAILQ_FIRST(&mq->mq_msgq)) == NULL && error == 0) {
1921 mtx_unlock(&mq->mq_mutex);
1924 mq->mq_receivers++;
1925 error = msleep(&mq->mq_receivers, &mq->mq_mutex,
1927 mq->mq_receivers--;
1933 TAILQ_REMOVE(&mq->mq_msgq, *msg, msg_link);
1934 mq->mq_curmsgs--;
1935 mq->mq_totalbytes -= (*msg)->msg_size;
1936 if (mq->mq_senders)
1937 wakeup_one(&mq->mq_senders);
1938 if (mq->mq_flags & MQ_WSEL) {
1939 mq->mq_flags &= ~MQ_WSEL;
1940 selwakeup(&mq->mq_wsel);
1942 KNOTE_LOCKED(&mq->mq_wsel.si_note, 0);
1944 if (mq->mq_notifier != NULL && mq->mq_receivers == 0 &&
1945 !TAILQ_EMPTY(&mq->mq_msgq)) {
1946 mqueue_send_notification(mq);
1948 mtx_unlock(&mq->mq_mutex);
1990 notifier_remove(struct proc *p, struct mqueue *mq, int fd)
1994 mtx_assert(&mq->mq_mutex, MA_OWNED);
1998 if (mq->mq_notifier == nt)
1999 mq->mq_notifier = NULL;
2014 struct mqueue *mq;
2022 mq = NULL;
2065 mq = mqueue_alloc(attr);
2066 if (mq == NULL) {
2074 mqueue_free(mq);
2080 pn->mn_data = mq;
2230 struct mqueue *mq;
2238 error = getmq(td, mqd, &fp, NULL, &mq);
2241 oattr->mq_maxmsg = mq->mq_maxmsg;
2242 oattr->mq_msgsize = mq->mq_msgsize;
2243 oattr->mq_curmsgs = mq->mq_curmsgs;
2281 struct mqueue *mq;
2286 error = getmq_read(td, mqd, &fp, NULL, &mq);
2290 error = mqueue_receive(mq, msg_ptr, msg_len, msg_prio, waitok,
2318 struct mqueue *mq;
2323 error = getmq_write(td, mqd, &fp, NULL, &mq);
2327 error = mqueue_send(mq, msg_ptr, msg_len, msg_prio, waitok,
2356 struct mqueue *mq;
2374 error = getmq(td, mqd, &fp, NULL, &mq);
2397 mtx_lock(&mq->mq_mutex);
2400 if (mq->mq_notifier != NULL) {
2408 mtx_unlock(&mq->mq_mutex);
2431 mq->mq_notifier = nt;
2438 if (mq->mq_receivers == 0 &&
2439 !TAILQ_EMPTY(&mq->mq_msgq))
2440 mqueue_send_notification(mq);
2443 notifier_remove(p, mq, mqd);
2445 mtx_unlock(&mq->mq_mutex);
2474 struct mqueue *mq;
2483 mq = FPTOMQ(fp);
2484 mtx_lock(&mq->mq_mutex);
2485 notifier_remove(td->td_proc, mq, fd);
2488 if (mq->mq_flags & MQ_RSEL) {
2489 mq->mq_flags &= ~MQ_RSEL;
2490 selwakeup(&mq->mq_rsel);
2492 if (mq->mq_flags & MQ_WSEL) {
2493 mq->mq_flags &= ~MQ_WSEL;
2494 selwakeup(&mq->mq_wsel);
2496 mtx_unlock(&mq->mq_mutex);
2505 struct mqueue *mq;
2513 mq = FPTOMQ(fp);
2514 mtx_lock(&mq->mq_mutex);
2516 mtx_unlock(&mq->mq_mutex);
2520 KASSERT(LIST_EMPTY(&p->p_mqnotifier), ("mq notifiers left"));
2527 struct mqueue *mq = FPTOMQ(fp);
2530 mtx_lock(&mq->mq_mutex);
2532 if (mq->mq_curmsgs) {
2535 mq->mq_flags |= MQ_RSEL;
2536 selrecord(td, &mq->mq_rsel);
2540 if (mq->mq_curmsgs < mq->mq_maxmsg)
2543 mq->mq_flags |= MQ_WSEL;
2544 selrecord(td, &mq->mq_wsel);
2547 mtx_unlock(&mq->mq_mutex);
2631 struct mqueue *mq = FPTOMQ(fp);
2636 knlist_add(&mq->mq_rsel.si_note, kn, 0);
2639 knlist_add(&mq->mq_wsel.si_note, kn, 0);
2648 struct mqueue *mq = FPTOMQ(kn->kn_fp);
2651 knlist_remove(&mq->mq_rsel.si_note, kn, 0);
2653 knlist_remove(&mq->mq_wsel.si_note, kn, 0);
2661 struct mqueue *mq = FPTOMQ(kn->kn_fp);
2663 mtx_assert(&mq->mq_mutex, MA_OWNED);
2664 return (mq->mq_curmsgs != 0);
2670 struct mqueue *mq = FPTOMQ(kn->kn_fp);
2672 mtx_assert(&mq->mq_mutex, MA_OWNED);
2673 return (mq->mq_curmsgs < mq->mq_maxmsg);