/src/rtpproxy/src/rtpp_sessinfo.c

Source (jump to first uncovered line)
/*
 * Copyright (c) 2004-2006 Maxim Sobolev <sobomax@FreeBSD.org>
 * Copyright (c) 2006-2015 Sippy Software, Inc., http://www.sippysoft.com
 * All rights reserved.
 *
 * 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.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``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 OR CONTRIBUTORS 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 "config.h"

#include <sys/stat.h>
#include <assert.h>
#include <stddef.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <poll.h>
#include <pthread.h>
#include <unistd.h>

#include "rtpp_types.h"
#include "rtpp_codeptr.h"
#include "rtpp_refcnt.h"
#include "rtpp_cfg.h"
#include "rtpp_sessinfo.h"
#include "rtpp_sessinfo_fin.h"
#include "rtpp_pipe.h"
#include "rtpp_stream.h"
#include "rtpp_session.h"
#include "rtpp_socket.h"
#include "rtpp_mallocs.h"
#include "rtpp_epoll.h"
#include "rtpp_debug.h"

enum polltbl_hst_ops {HST_ADD, HST_DEL, HST_UPD};

struct rtpp_polltbl_hst_ent {
   uint64_t stuid;
   enum polltbl_hst_ops op;
   struct rtpp_socket *skt;
};

struct rtpp_polltbl_hst_part {
   int alen;  /* Number of entries allocated */
   struct rtpp_polltbl_hst_ent *clog;
};

struct rtpp_polltbl_hst {
   int ulen;  /* Number of entries used */
   int ilen;  /* Minimum number of entries to be allocated when need to extend */
   struct rtpp_polltbl_hst_part main;
   struct rtpp_polltbl_hst_part shadow;
   struct rtpp_weakref *streams_wrt;
   pthread_mutex_t lock;
};

struct rtpp_sessinfo_priv {
   struct rtpp_sessinfo pub;
   struct rtpp_polltbl_hst hst_rtp;
   struct rtpp_polltbl_hst hst_rtcp;
};

static int rtpp_sinfo_append(struct rtpp_sessinfo *, struct rtpp_session *,
  int, struct rtpp_socket **);
static void rtpp_sinfo_update(struct rtpp_sessinfo *, struct rtpp_session *,
  int, struct rtpp_socket **);
static void rtpp_sinfo_remove(struct rtpp_sessinfo *, struct rtpp_session *,
  int);
static int rtpp_sinfo_sync_polltbl(struct rtpp_sessinfo *, struct rtpp_polltbl *,
  int);
static void rtpp_sessinfo_dtor(struct rtpp_sessinfo_priv *);

DEFINE_SMETHODS(rtpp_sessinfo,
    .append = &rtpp_sinfo_append,
    .update = &rtpp_sinfo_update,
    .remove = &rtpp_sinfo_remove,
    .sync_polltbl = &rtpp_sinfo_sync_polltbl,
);

static int
rtpp_polltbl_hst_alloc(struct rtpp_polltbl_hst *hp, int alen)
{

    hp->main.clog = rtpp_zmalloc(sizeof(struct rtpp_polltbl_hst_ent) * alen);
    if (hp->main.clog == NULL) {
        goto e0;
    }
    hp->shadow.clog = rtpp_zmalloc(sizeof(struct rtpp_polltbl_hst_ent) * alen);
    if (hp->shadow.clog == NULL) {
        goto e1;
    }
    if (pthread_mutex_init(&hp->lock, NULL) != 0)
        goto e2;
    hp->main.alen = hp->shadow.alen = hp->ilen = alen;
    return (0);
e2:
    free(hp->shadow.clog);
e1:
    free(hp->main.clog);
e0:
    return (-1);
}

static void
rtpp_polltbl_hst_dtor(struct rtpp_polltbl_hst *hp)
{
    int i;
    struct rtpp_polltbl_hst_ent *hep;

    for (i = 0; i < hp->ulen; i++) {
        hep = hp->main.clog + i;
        if (hep->skt != NULL) {
            RTPP_OBJ_DECREF(hep->skt);
        }
    }
    if (hp->main.alen > 0) {
        free(hp->shadow.clog);
        free(hp->main.clog);
        pthread_mutex_destroy(&hp->lock);
    }
}

static int
rtpp_polltbl_hst_extend(struct rtpp_polltbl_hst *hp)
{
    struct rtpp_polltbl_hst_ent *clog_new;
    size_t alen = sizeof(struct rtpp_polltbl_hst_ent) *
      (hp->main.alen + hp->ilen);

    clog_new = realloc(hp->main.clog, alen);
    if (clog_new == NULL) {
        return (-1);
    }
    hp->main.clog = clog_new;
    hp->main.alen += hp->ilen;
    return (0);
}

static void
rtpp_polltbl_hst_record(struct rtpp_polltbl_hst *hp, enum polltbl_hst_ops op,
  uint64_t stuid, struct rtpp_socket *skt)
{
    struct rtpp_polltbl_hst_ent *hpe;

    hpe = hp->main.clog + hp->ulen;
    hpe->op = op;
    hpe->stuid = stuid;
    hpe->skt = skt;
    hp->ulen += 1;
    if (skt != NULL) {
        RTPP_OBJ_INCREF(skt);
    }
}

struct rtpp_sessinfo *
rtpp_sessinfo_ctor(const struct rtpp_cfg *cfsp)
{
    struct rtpp_sessinfo *sessinfo;
    struct rtpp_sessinfo_priv *pvt;

    pvt = rtpp_rzmalloc(sizeof(struct rtpp_sessinfo_priv), PVT_RCOFFS(pvt));
    if (pvt == NULL) {
        return (NULL);
    }
    sessinfo = &(pvt->pub);
    if (rtpp_polltbl_hst_alloc(&pvt->hst_rtp, 10) != 0) {
        goto e6;
    }
    if (rtpp_polltbl_hst_alloc(&pvt->hst_rtcp, 10) != 0) {
        goto e7;
    }
    pvt->hst_rtp.streams_wrt = cfsp->rtp_streams_wrt;
    pvt->hst_rtcp.streams_wrt = cfsp->rtcp_streams_wrt;

    PUBINST_FININIT(&pvt->pub, pvt, rtpp_sessinfo_dtor);
    return (sessinfo);

e7:
    rtpp_polltbl_hst_dtor(&pvt->hst_rtp);
e6:
    RTPP_OBJ_DECREF(&(pvt->pub));
    return (NULL);
}

static void
rtpp_sessinfo_dtor(struct rtpp_sessinfo_priv *pvt)
{

    rtpp_sessinfo_fin(&(pvt->pub));
    rtpp_polltbl_hst_dtor(&pvt->hst_rtp);
    rtpp_polltbl_hst_dtor(&pvt->hst_rtcp);
}

static int
rtpp_sinfo_append(struct rtpp_sessinfo *sessinfo, struct rtpp_session *sp,
  int index, struct rtpp_socket **new_fds)
{
    struct rtpp_sessinfo_priv *pvt;
    struct rtpp_stream *rtp, *rtcp;

    PUB2PVT(sessinfo, pvt);
    pthread_mutex_lock(&pvt->hst_rtp.lock);
    if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
        if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
            goto e0;
        }
    }
    pthread_mutex_lock(&pvt->hst_rtcp.lock);
    if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
        if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
            goto e1;
        }
    }
    rtp = sp->rtp->stream[index];
    CALL_SMETHOD(rtp, set_skt, new_fds[0]);
    rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_ADD, rtp->stuid, new_fds[0]);
    pthread_mutex_unlock(&pvt->hst_rtp.lock);

    rtcp = sp->rtcp->stream[index];
    CALL_SMETHOD(rtcp, set_skt, new_fds[1]);
    rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_ADD, rtcp->stuid, new_fds[1]);
    pthread_mutex_unlock(&pvt->hst_rtcp.lock);

    return (0);
e1:
    pthread_mutex_unlock(&pvt->hst_rtcp.lock);
e0:
    pthread_mutex_unlock(&pvt->hst_rtp.lock);
    return (-1);
}

static int
find_polltbl_idx(struct rtpp_polltbl *ptp, uint64_t stuid)
{
    int i, j = -1;

    for (i = 0; i < ptp->curlen; i++) {
        if (ptp->mds[i].stuid != stuid)
            continue;
        RTPP_DBGCODE() {
            assert(j == -1);
            j = i;
        } else {
            return (i);
        }
    }
    return (j);
}

static void
rtpp_sinfo_update(struct rtpp_sessinfo *sessinfo, struct rtpp_session *sp,
  int index, struct rtpp_socket **new_fds)
{
    struct rtpp_sessinfo_priv *pvt;
    struct rtpp_stream *rtp, *rtcp;
    struct rtpp_socket *old_fd;

    PUB2PVT(sessinfo, pvt);

    pthread_mutex_lock(&pvt->hst_rtp.lock);
    if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
        if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
            goto e0;
        }
    }
    pthread_mutex_lock(&pvt->hst_rtcp.lock);
    if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
        if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
            goto e1;
        }
    }
    rtp = sp->rtp->stream[index];
    old_fd = CALL_SMETHOD(rtp, update_skt, new_fds[0]);
    if (old_fd != NULL) {
        rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_UPD, rtp->stuid, new_fds[0]);
        pthread_mutex_unlock(&pvt->hst_rtp.lock);
        RTPP_OBJ_DECREF(old_fd);
    } else {
        rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_ADD, rtp->stuid, new_fds[0]);
        pthread_mutex_unlock(&pvt->hst_rtp.lock);
    }
    rtcp = sp->rtcp->stream[index];
    old_fd = CALL_SMETHOD(rtcp, update_skt, new_fds[1]);
    if (old_fd != NULL) {
        rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_UPD, rtcp->stuid, new_fds[1]);
        pthread_mutex_unlock(&pvt->hst_rtcp.lock);
        RTPP_OBJ_DECREF(old_fd);
    } else {
        rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_ADD, rtcp->stuid, new_fds[1]);
        pthread_mutex_unlock(&pvt->hst_rtcp.lock);
    }
    return;
e1:
    pthread_mutex_unlock(&pvt->hst_rtcp.lock);
e0:
    pthread_mutex_unlock(&pvt->hst_rtp.lock);
}

static void
rtpp_sinfo_remove(struct rtpp_sessinfo *sessinfo, struct rtpp_session *sp,
  int index)
{
    struct rtpp_sessinfo_priv *pvt;
    struct rtpp_stream *rtp, *rtcp;
    struct rtpp_socket *fd_rtp, *fd_rtcp;;

    PUB2PVT(sessinfo, pvt);

    rtp = sp->rtp->stream[index];
    rtcp = sp->rtcp->stream[index];
    fd_rtp = CALL_SMETHOD(rtp, get_skt, HEREVAL);
    fd_rtcp = CALL_SMETHOD(rtcp, get_skt, HEREVAL);
    if (fd_rtp != NULL) {
        pthread_mutex_lock(&pvt->hst_rtp.lock);
        if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
            if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
                goto e0;
            }
        }
    }
    if (fd_rtcp != NULL) {
        pthread_mutex_lock(&pvt->hst_rtcp.lock);
        if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
            if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
                goto e1;
            }
        }
    }
    if (fd_rtp != NULL) {
        rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_DEL, rtp->stuid, NULL);
        pthread_mutex_unlock(&pvt->hst_rtp.lock);
    }
    if (fd_rtcp != NULL) {
        rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_DEL, rtcp->stuid, NULL);
        pthread_mutex_unlock(&pvt->hst_rtcp.lock);
    }
    if (fd_rtp != NULL)
        RTPP_OBJ_DECREF(fd_rtp);
    if (fd_rtcp != NULL)
        RTPP_OBJ_DECREF(fd_rtcp);
    return;
e1:
    pthread_mutex_unlock(&pvt->hst_rtcp.lock);
e0:
    if (fd_rtp != NULL)
        pthread_mutex_unlock(&pvt->hst_rtp.lock);
}

void
rtpp_polltbl_free(struct rtpp_polltbl *ptbl)
{
    int i;

    if (ptbl->aloclen != 0) {
        for (i = 0; i < ptbl->curlen; i++) {
            int fd = CALL_SMETHOD(ptbl->mds[i].skt, getfd);
            rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, fd, NULL);
            RTPP_OBJ_DECREF(ptbl->mds[i].skt);
        }
        free(ptbl->mds);
    }
    close(ptbl->wakefd[0]);
    close(ptbl->epfd);
}

static int
rtpp_sinfo_sync_polltbl(struct rtpp_sessinfo *sessinfo,
  struct rtpp_polltbl *ptbl, int pipe_type)
{
    struct rtpp_sessinfo_priv *pvt;
    struct rtpp_polltbl_mdata *mds;
    struct rtpp_polltbl_hst *hp;
    struct rtpp_polltbl_hst_ent *clog;
    int i, ulen;

    PUB2PVT(sessinfo, pvt);

    hp = (pipe_type == PIPE_RTP) ? &pvt->hst_rtp : &pvt->hst_rtcp;

    pthread_mutex_lock(&hp->lock);
    if (hp->ulen == 0) {
        pthread_mutex_unlock(&hp->lock);
        return (0);
    }

    if (hp->ulen > ptbl->aloclen - ptbl->curlen) {
        int alen = hp->ulen + ptbl->curlen;

        mds = realloc(ptbl->mds, (alen * sizeof(struct rtpp_polltbl_mdata)));
        if (mds == NULL) {
            goto e0;
        }
        ptbl->mds = mds;
        ptbl->aloclen = alen;
    }

    struct rtpp_polltbl_hst_part hpp = hp->main;
    hp->main = hp->shadow;
    hp->shadow = hpp;
    clog = hpp.clog;
    ulen = hp->ulen;
    hp->ulen = 0;
    ptbl->streams_wrt = hp->streams_wrt;
    pthread_mutex_unlock(&hp->lock);

    for (i = 0; i < ulen; i++) {
        struct rtpp_polltbl_hst_ent *hep;
        int session_index, movelen;
        struct epoll_event event;

        hep = clog + i;
        switch (hep->op) {
        case HST_ADD:
#ifdef RTPP_DEBUG
            assert(find_polltbl_idx(ptbl, hep->stuid) < 0);
#endif
            session_index = ptbl->curlen;
            event.events = EPOLLIN;
            event.data.ptr = hep->skt;
            rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_ADD, CALL_SMETHOD(hep->skt, getfd), &event);
            ptbl->mds[session_index].stuid = hep->stuid;
            ptbl->mds[session_index].skt = hep->skt;
            ptbl->curlen++;
            break;

        case HST_DEL:
            session_index = find_polltbl_idx(ptbl, hep->stuid);
            assert(session_index > -1);
            rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, CALL_SMETHOD(ptbl->mds[session_index].skt, getfd), NULL);
            RTPP_OBJ_DECREF(ptbl->mds[session_index].skt);
            movelen = (ptbl->curlen - session_index - 1);
            if (movelen > 0) {
                memmove(&ptbl->mds[session_index], &ptbl->mds[session_index + 1],
                  movelen * sizeof(ptbl->mds[0]));
            }
            ptbl->curlen--;
            break;

        case HST_UPD:
            session_index = find_polltbl_idx(ptbl, hep->stuid);
            assert(session_index > -1);
            rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, CALL_SMETHOD(ptbl->mds[session_index].skt, getfd), NULL);
            RTPP_OBJ_DECREF(ptbl->mds[session_index].skt);
            event.events = EPOLLIN;
            event.data.ptr = hep->skt;
            rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_ADD, CALL_SMETHOD(hep->skt, getfd), &event);
            ptbl->mds[session_index].skt = hep->skt;
            break;
        }
    }
    ptbl->revision += ulen;

    return (1);
e0:
    for (i = 0; i < hp->ulen; i++) {
        struct rtpp_polltbl_hst_ent *hep;

        hep = hp->main.clog + i;
        if (hep->skt != NULL) {
            RTPP_OBJ_DECREF(hep->skt);
        }
    }
    hp->ulen = 0;
    pthread_mutex_unlock(&hp->lock);
    return (-1);
}

Coverage Report

Created: 2025-08-08 07:04

Line	Count	Source (jump to first uncovered line)
1		/*
2		* Copyright (c) 2004-2006 Maxim Sobolev <sobomax@FreeBSD.org>
3		* Copyright (c) 2006-2015 Sippy Software, Inc., http://www.sippysoft.com
4		* All rights reserved.
5		*
6		* Redistribution and use in source and binary forms, with or without
7		* modification, are permitted provided that the following conditions
8		* are met:
9		* 1. Redistributions of source code must retain the above copyright
10		* notice, this list of conditions and the following disclaimer.
11		* 2. Redistributions in binary form must reproduce the above copyright
12		* notice, this list of conditions and the following disclaimer in the
13		* documentation and/or other materials provided with the distribution.
14		*
15		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
16		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
17		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
18		* ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
19		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
20		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
21		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
22		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
23		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
24		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
25		* SUCH DAMAGE.
26		*
27		*/
28
29		#include "config.h"
30
31		#include <sys/stat.h>
32		#include <assert.h>
33		#include <stddef.h>
34		#include <stdint.h>
35		#include <stdlib.h>
36		#include <string.h>
37		#include <poll.h>
38		#include <pthread.h>
39		#include <unistd.h>
40
41		#include "rtpp_types.h"
42		#include "rtpp_codeptr.h"
43		#include "rtpp_refcnt.h"
44		#include "rtpp_cfg.h"
45		#include "rtpp_sessinfo.h"
46		#include "rtpp_sessinfo_fin.h"
47		#include "rtpp_pipe.h"
48		#include "rtpp_stream.h"
49		#include "rtpp_session.h"
50		#include "rtpp_socket.h"
51		#include "rtpp_mallocs.h"
52		#include "rtpp_epoll.h"
53		#include "rtpp_debug.h"
54
55		enum polltbl_hst_ops {HST_ADD, HST_DEL, HST_UPD};
56
57		struct rtpp_polltbl_hst_ent {
58		uint64_t stuid;
59		enum polltbl_hst_ops op;
60		struct rtpp_socket *skt;
61		};
62
63		struct rtpp_polltbl_hst_part {
64		int alen; /* Number of entries allocated */
65		struct rtpp_polltbl_hst_ent *clog;
66		};
67
68		struct rtpp_polltbl_hst {
69		int ulen; /* Number of entries used */
70		int ilen; /* Minimum number of entries to be allocated when need to extend */
71		struct rtpp_polltbl_hst_part main;
72		struct rtpp_polltbl_hst_part shadow;
73		struct rtpp_weakref *streams_wrt;
74		pthread_mutex_t lock;
75		};
76
77		struct rtpp_sessinfo_priv {
78		struct rtpp_sessinfo pub;
79		struct rtpp_polltbl_hst hst_rtp;
80		struct rtpp_polltbl_hst hst_rtcp;
81		};
82
83		static int rtpp_sinfo_append(struct rtpp_sessinfo , struct rtpp_session ,
84		int, struct rtpp_socket **);
85		static void rtpp_sinfo_update(struct rtpp_sessinfo , struct rtpp_session ,
86		int, struct rtpp_socket **);
87		static void rtpp_sinfo_remove(struct rtpp_sessinfo , struct rtpp_session ,
88		int);
89		static int rtpp_sinfo_sync_polltbl(struct rtpp_sessinfo , struct rtpp_polltbl ,
90		int);
91		static void rtpp_sessinfo_dtor(struct rtpp_sessinfo_priv *);
92
93		DEFINE_SMETHODS(rtpp_sessinfo,
94		.append = &rtpp_sinfo_append,
95		.update = &rtpp_sinfo_update,
96		.remove = &rtpp_sinfo_remove,
97		.sync_polltbl = &rtpp_sinfo_sync_polltbl,
98		);
99
100		static int
101		rtpp_polltbl_hst_alloc(struct rtpp_polltbl_hst *hp, int alen)
102	0	{
103
104	0	hp->main.clog = rtpp_zmalloc(sizeof(struct rtpp_polltbl_hst_ent) * alen);
105	0	if (hp->main.clog == NULL) {
106	0	goto e0;
107	0	}
108	0	hp->shadow.clog = rtpp_zmalloc(sizeof(struct rtpp_polltbl_hst_ent) * alen);
109	0	if (hp->shadow.clog == NULL) {
110	0	goto e1;
111	0	}
112	0	if (pthread_mutex_init(&hp->lock, NULL) != 0)
113	0	goto e2;
114	0	hp->main.alen = hp->shadow.alen = hp->ilen = alen;
115	0	return (0);
116	0	e2:
117	0	free(hp->shadow.clog);
118	0	e1:
119	0	free(hp->main.clog);
120	0	e0:
121	0	return (-1);
122	0	}
123
124		static void
125		rtpp_polltbl_hst_dtor(struct rtpp_polltbl_hst *hp)
126	0	{
127	0	int i;
128	0	struct rtpp_polltbl_hst_ent *hep;
129
130	0	for (i = 0; i < hp->ulen; i++) {
131	0	hep = hp->main.clog + i;
132	0	if (hep->skt != NULL) {
133	0	RTPP_OBJ_DECREF(hep->skt);
134	0	}
135	0	}
136	0	if (hp->main.alen > 0) {
137	0	free(hp->shadow.clog);
138	0	free(hp->main.clog);
139	0	pthread_mutex_destroy(&hp->lock);
140	0	}
141	0	}
142
143		static int
144		rtpp_polltbl_hst_extend(struct rtpp_polltbl_hst *hp)
145	0	{
146	0	struct rtpp_polltbl_hst_ent *clog_new;
147	0	size_t alen = sizeof(struct rtpp_polltbl_hst_ent) *
148	0	(hp->main.alen + hp->ilen);
149
150	0	clog_new = realloc(hp->main.clog, alen);
151	0	if (clog_new == NULL) {
152	0	return (-1);
153	0	}
154	0	hp->main.clog = clog_new;
155	0	hp->main.alen += hp->ilen;
156	0	return (0);
157	0	}
158
159		static void
160		rtpp_polltbl_hst_record(struct rtpp_polltbl_hst *hp, enum polltbl_hst_ops op,
161		uint64_t stuid, struct rtpp_socket *skt)
162	0	{
163	0	struct rtpp_polltbl_hst_ent *hpe;
164
165	0	hpe = hp->main.clog + hp->ulen;
166	0	hpe->op = op;
167	0	hpe->stuid = stuid;
168	0	hpe->skt = skt;
169	0	hp->ulen += 1;
170	0	if (skt != NULL) {
171	0	RTPP_OBJ_INCREF(skt);
172	0	}
173	0	}
174
175		struct rtpp_sessinfo *
176		rtpp_sessinfo_ctor(const struct rtpp_cfg *cfsp)
177	0	{
178	0	struct rtpp_sessinfo *sessinfo;
179	0	struct rtpp_sessinfo_priv *pvt;
180
181	0	pvt = rtpp_rzmalloc(sizeof(struct rtpp_sessinfo_priv), PVT_RCOFFS(pvt));
182	0	if (pvt == NULL) {
183	0	return (NULL);
184	0	}
185	0	sessinfo = &(pvt->pub);
186	0	if (rtpp_polltbl_hst_alloc(&pvt->hst_rtp, 10) != 0) {
187	0	goto e6;
188	0	}
189	0	if (rtpp_polltbl_hst_alloc(&pvt->hst_rtcp, 10) != 0) {
190	0	goto e7;
191	0	}
192	0	pvt->hst_rtp.streams_wrt = cfsp->rtp_streams_wrt;
193	0	pvt->hst_rtcp.streams_wrt = cfsp->rtcp_streams_wrt;
194
195	0	PUBINST_FININIT(&pvt->pub, pvt, rtpp_sessinfo_dtor);
196	0	return (sessinfo);
197
198	0	e7:
199	0	rtpp_polltbl_hst_dtor(&pvt->hst_rtp);
200	0	e6:
201	0	RTPP_OBJ_DECREF(&(pvt->pub));
202	0	return (NULL);
203	0	}
204
205		static void
206		rtpp_sessinfo_dtor(struct rtpp_sessinfo_priv *pvt)
207	0	{
208
209	0	rtpp_sessinfo_fin(&(pvt->pub));
210	0	rtpp_polltbl_hst_dtor(&pvt->hst_rtp);
211	0	rtpp_polltbl_hst_dtor(&pvt->hst_rtcp);
212	0	}
213
214		static int
215		rtpp_sinfo_append(struct rtpp_sessinfo sessinfo, struct rtpp_session sp,
216		int index, struct rtpp_socket **new_fds)
217	0	{
218	0	struct rtpp_sessinfo_priv *pvt;
219	0	struct rtpp_stream rtp, rtcp;
220
221	0	PUB2PVT(sessinfo, pvt);
222	0	pthread_mutex_lock(&pvt->hst_rtp.lock);
223	0	if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
224	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
225	0	goto e0;
226	0	}
227	0	}
228	0	pthread_mutex_lock(&pvt->hst_rtcp.lock);
229	0	if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
230	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
231	0	goto e1;
232	0	}
233	0	}
234	0	rtp = sp->rtp->stream[index];
235	0	CALL_SMETHOD(rtp, set_skt, new_fds[0]);
236	0	rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_ADD, rtp->stuid, new_fds[0]);
237	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
238
239	0	rtcp = sp->rtcp->stream[index];
240	0	CALL_SMETHOD(rtcp, set_skt, new_fds[1]);
241	0	rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_ADD, rtcp->stuid, new_fds[1]);
242	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
243
244	0	return (0);
245	0	e1:
246	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
247	0	e0:
248	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
249	0	return (-1);
250	0	}
251
252		static int
253		find_polltbl_idx(struct rtpp_polltbl *ptp, uint64_t stuid)
254	0	{
255	0	int i, j = -1;
256
257	0	for (i = 0; i < ptp->curlen; i++) {
258	0	if (ptp->mds[i].stuid != stuid)
259	0	continue;
260	0	RTPP_DBGCODE() {
261	0	assert(j == -1);
262	0	j = i;
263	0	} else {
264	0	return (i);
265	0	}
266	0	}
267	0	return (j);
268	0	}
269
270		static void
271		rtpp_sinfo_update(struct rtpp_sessinfo sessinfo, struct rtpp_session sp,
272		int index, struct rtpp_socket **new_fds)
273	0	{
274	0	struct rtpp_sessinfo_priv *pvt;
275	0	struct rtpp_stream rtp, rtcp;
276	0	struct rtpp_socket *old_fd;
277
278	0	PUB2PVT(sessinfo, pvt);
279
280	0	pthread_mutex_lock(&pvt->hst_rtp.lock);
281	0	if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
282	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
283	0	goto e0;
284	0	}
285	0	}
286	0	pthread_mutex_lock(&pvt->hst_rtcp.lock);
287	0	if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
288	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
289	0	goto e1;
290	0	}
291	0	}
292	0	rtp = sp->rtp->stream[index];
293	0	old_fd = CALL_SMETHOD(rtp, update_skt, new_fds[0]);
294	0	if (old_fd != NULL) {
295	0	rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_UPD, rtp->stuid, new_fds[0]);
296	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
297	0	RTPP_OBJ_DECREF(old_fd);
298	0	} else {
299	0	rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_ADD, rtp->stuid, new_fds[0]);
300	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
301	0	}
302	0	rtcp = sp->rtcp->stream[index];
303	0	old_fd = CALL_SMETHOD(rtcp, update_skt, new_fds[1]);
304	0	if (old_fd != NULL) {
305	0	rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_UPD, rtcp->stuid, new_fds[1]);
306	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
307	0	RTPP_OBJ_DECREF(old_fd);
308	0	} else {
309	0	rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_ADD, rtcp->stuid, new_fds[1]);
310	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
311	0	}
312	0	return;
313	0	e1:
314	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
315	0	e0:
316	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
317	0	}
318
319		static void
320		rtpp_sinfo_remove(struct rtpp_sessinfo sessinfo, struct rtpp_session sp,
321		int index)
322	0	{
323	0	struct rtpp_sessinfo_priv *pvt;
324	0	struct rtpp_stream rtp, rtcp;
325	0	struct rtpp_socket fd_rtp, fd_rtcp;;
326
327	0	PUB2PVT(sessinfo, pvt);
328
329	0	rtp = sp->rtp->stream[index];
330	0	rtcp = sp->rtcp->stream[index];
331	0	fd_rtp = CALL_SMETHOD(rtp, get_skt, HEREVAL);
332	0	fd_rtcp = CALL_SMETHOD(rtcp, get_skt, HEREVAL);
333	0	if (fd_rtp != NULL) {
334	0	pthread_mutex_lock(&pvt->hst_rtp.lock);
335	0	if (pvt->hst_rtp.ulen == pvt->hst_rtp.main.alen) {
336	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtp) < 0) {
337	0	goto e0;
338	0	}
339	0	}
340	0	}
341	0	if (fd_rtcp != NULL) {
342	0	pthread_mutex_lock(&pvt->hst_rtcp.lock);
343	0	if (pvt->hst_rtcp.ulen == pvt->hst_rtcp.main.alen) {
344	0	if (rtpp_polltbl_hst_extend(&pvt->hst_rtcp) < 0) {
345	0	goto e1;
346	0	}
347	0	}
348	0	}
349	0	if (fd_rtp != NULL) {
350	0	rtpp_polltbl_hst_record(&pvt->hst_rtp, HST_DEL, rtp->stuid, NULL);
351	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
352	0	}
353	0	if (fd_rtcp != NULL) {
354	0	rtpp_polltbl_hst_record(&pvt->hst_rtcp, HST_DEL, rtcp->stuid, NULL);
355	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
356	0	}
357	0	if (fd_rtp != NULL)
358	0	RTPP_OBJ_DECREF(fd_rtp);
359	0	if (fd_rtcp != NULL)
360	0	RTPP_OBJ_DECREF(fd_rtcp);
361	0	return;
362	0	e1:
363	0	pthread_mutex_unlock(&pvt->hst_rtcp.lock);
364	0	e0:
365	0	if (fd_rtp != NULL)
366	0	pthread_mutex_unlock(&pvt->hst_rtp.lock);
367	0	}
368
369		void
370		rtpp_polltbl_free(struct rtpp_polltbl *ptbl)
371	0	{
372	0	int i;
373
374	0	if (ptbl->aloclen != 0) {
375	0	for (i = 0; i < ptbl->curlen; i++) {
376	0	int fd = CALL_SMETHOD(ptbl->mds[i].skt, getfd);
377	0	rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, fd, NULL);
378	0	RTPP_OBJ_DECREF(ptbl->mds[i].skt);
379	0	}
380	0	free(ptbl->mds);
381	0	}
382	0	close(ptbl->wakefd[0]);
383	0	close(ptbl->epfd);
384	0	}
385
386		static int
387		rtpp_sinfo_sync_polltbl(struct rtpp_sessinfo *sessinfo,
388		struct rtpp_polltbl *ptbl, int pipe_type)
389	0	{
390	0	struct rtpp_sessinfo_priv *pvt;
391	0	struct rtpp_polltbl_mdata *mds;
392	0	struct rtpp_polltbl_hst *hp;
393	0	struct rtpp_polltbl_hst_ent *clog;
394	0	int i, ulen;
395
396	0	PUB2PVT(sessinfo, pvt);
397
398	0	hp = (pipe_type == PIPE_RTP) ? &pvt->hst_rtp : &pvt->hst_rtcp;
399
400	0	pthread_mutex_lock(&hp->lock);
401	0	if (hp->ulen == 0) {
402	0	pthread_mutex_unlock(&hp->lock);
403	0	return (0);
404	0	}
405
406	0	if (hp->ulen > ptbl->aloclen - ptbl->curlen) {
407	0	int alen = hp->ulen + ptbl->curlen;
408
409	0	mds = realloc(ptbl->mds, (alen * sizeof(struct rtpp_polltbl_mdata)));
410	0	if (mds == NULL) {
411	0	goto e0;
412	0	}
413	0	ptbl->mds = mds;
414	0	ptbl->aloclen = alen;
415	0	}
416
417	0	struct rtpp_polltbl_hst_part hpp = hp->main;
418	0	hp->main = hp->shadow;
419	0	hp->shadow = hpp;
420	0	clog = hpp.clog;
421	0	ulen = hp->ulen;
422	0	hp->ulen = 0;
423	0	ptbl->streams_wrt = hp->streams_wrt;
424	0	pthread_mutex_unlock(&hp->lock);
425
426	0	for (i = 0; i < ulen; i++) {
427	0	struct rtpp_polltbl_hst_ent *hep;
428	0	int session_index, movelen;
429	0	struct epoll_event event;
430
431	0	hep = clog + i;
432	0	switch (hep->op) {
433	0	case HST_ADD:
434		#ifdef RTPP_DEBUG
435		assert(find_polltbl_idx(ptbl, hep->stuid) < 0);
436		#endif
437	0	session_index = ptbl->curlen;
438	0	event.events = EPOLLIN;
439	0	event.data.ptr = hep->skt;
440	0	rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_ADD, CALL_SMETHOD(hep->skt, getfd), &event);
441	0	ptbl->mds[session_index].stuid = hep->stuid;
442	0	ptbl->mds[session_index].skt = hep->skt;
443	0	ptbl->curlen++;
444	0	break;
445
446	0	case HST_DEL:
447	0	session_index = find_polltbl_idx(ptbl, hep->stuid);
448	0	assert(session_index > -1);
449	0	rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, CALL_SMETHOD(ptbl->mds[session_index].skt, getfd), NULL);
450	0	RTPP_OBJ_DECREF(ptbl->mds[session_index].skt);
451	0	movelen = (ptbl->curlen - session_index - 1);
452	0	if (movelen > 0) {
453	0	memmove(&ptbl->mds[session_index], &ptbl->mds[session_index + 1],
454	0	movelen * sizeof(ptbl->mds[0]));
455	0	}
456	0	ptbl->curlen--;
457	0	break;
458
459	0	case HST_UPD:
460	0	session_index = find_polltbl_idx(ptbl, hep->stuid);
461	0	assert(session_index > -1);
462	0	rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_DEL, CALL_SMETHOD(ptbl->mds[session_index].skt, getfd), NULL);
463	0	RTPP_OBJ_DECREF(ptbl->mds[session_index].skt);
464	0	event.events = EPOLLIN;
465	0	event.data.ptr = hep->skt;
466	0	rtpp_epoll_ctl(ptbl->epfd, EPOLL_CTL_ADD, CALL_SMETHOD(hep->skt, getfd), &event);
467	0	ptbl->mds[session_index].skt = hep->skt;
468	0	break;
469	0	}
470	0	}
471	0	ptbl->revision += ulen;
472
473	0	return (1);
474	0	e0:
475	0	for (i = 0; i < hp->ulen; i++) {
476	0	struct rtpp_polltbl_hst_ent *hep;
477
478	0	hep = hp->main.clog + i;
479	0	if (hep->skt != NULL) {
480	0	RTPP_OBJ_DECREF(hep->skt);
481	0	}
482	0	}
483	0	hp->ulen = 0;
484	0	pthread_mutex_unlock(&hp->lock);
485	0	return (-1);
486	0	}