frodovdr
/
vdr-plugin-iptv
mirror of https://github.com/rofafor/vdr-plugin-iptv.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Try to read all data from udp socket at once 

Existing implementation of udp read mechanism was giving up too easily.
Replaced it with version that tries more actively to drain the entire socket buffer.

Signed-off-by: Antti Seppälä <a.seppala@gmail.com>
This commit is contained in:
Antti Seppälä 2012-06-02 14:24:29 +03:00 committed by Rolf Ahrenberg
parent c7cbde301b
commit f4126b7e2c
1 changed files with 67 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
socket.c
View File

@ -154,70 +154,74 @@ int cIptvUdpSocket::Read(unsigned char* BufferAddr, unsigned int BufferLen)
error("Invalid socket in %s\n", __FUNCTION__); error("Invalid socket in %s\n", __FUNCTION__);
return -1; return -1;
} }
socklen_t addrlen = sizeof(sockAddr);
int len = 0; int len = 0;
struct msghdr msgh; // Read data from socket in a loop
struct cmsghdr *cmsg; do {
struct iovec iov; socklen_t addrlen = sizeof(sockAddr);
char cbuf[256]; struct msghdr msgh;
// Initialize iov and msgh structures struct cmsghdr *cmsg;
memset(&msgh, 0, sizeof(struct msghdr)); struct iovec iov;
iov.iov_base = BufferAddr; char cbuf[256];
iov.iov_len = BufferLen; len = 0;
msgh.msg_control = cbuf; // Initialize iov and msgh structures
msgh.msg_controllen = sizeof(cbuf); memset(&msgh, 0, sizeof(struct msghdr));
msgh.msg_name = &sockAddr; iov.iov_base = BufferAddr;
msgh.msg_namelen = addrlen; iov.iov_len = BufferLen;
msgh.msg_iov = &iov; msgh.msg_control = cbuf;
msgh.msg_iovlen = 1; msgh.msg_controllen = sizeof(cbuf);
msgh.msg_flags = 0; msgh.msg_name = &sockAddr;
// Read data from socket msgh.msg_namelen = addrlen;
if (isActive && socketDesc && BufferAddr && (BufferLen > 0)) msgh.msg_iov = &iov;
len = (int)recvmsg(socketDesc, &msgh, MSG_DONTWAIT); msgh.msg_iovlen = 1;
if (len < 0) { msgh.msg_flags = 0;
ERROR_IF(errno != EAGAIN, "recvmsg()");
return -1; if (isActive && socketDesc && BufferAddr && (BufferLen > 0))
} len = (int)recvmsg(socketDesc, &msgh, MSG_DONTWAIT);
else if (len > 0) { else
// Process auxiliary received data and validate source address break;
for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL; cmsg = CMSG_NXTHDR(&msgh, cmsg)) { if (len > 0) {
if ((cmsg->cmsg_level == SOL_IP) && (cmsg->cmsg_type == IP_PKTINFO)) { // Process auxiliary received data and validate source address
struct in_pktinfo *i = (struct in_pktinfo *)CMSG_DATA(cmsg); for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL; cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
if ((i->ipi_addr.s_addr == streamAddr) || (INADDR_ANY == streamAddr)) { if ((cmsg->cmsg_level == SOL_IP) && (cmsg->cmsg_type == IP_PKTINFO)) {
if (BufferAddr[0] == TS_SYNC_BYTE) struct in_pktinfo *i = (struct in_pktinfo *)CMSG_DATA(cmsg);
return len; if ((i->ipi_addr.s_addr == streamAddr) || (INADDR_ANY == streamAddr)) {
else if (len > 3) { if (BufferAddr[0] == TS_SYNC_BYTE)
// http://www.networksorcery.com/enp/rfc/rfc2250.txt return len;
// version else if (len > 3) {
unsigned int v = (BufferAddr[0] >> 6) & 0x03; // http://www.networksorcery.com/enp/rfc/rfc2250.txt
// extension bit // version
unsigned int x = (BufferAddr[0] >> 4) & 0x01; unsigned int v = (BufferAddr[0] >> 6) & 0x03;
// cscr count // extension bit
unsigned int cc = BufferAddr[0] & 0x0F; unsigned int x = (BufferAddr[0] >> 4) & 0x01;
// payload type: MPEG2 TS = 33 // cscr count
//unsigned int pt = readBuffer[1] & 0x7F; unsigned int cc = BufferAddr[0] & 0x0F;
// header lenght // payload type: MPEG2 TS = 33
unsigned int headerlen = (3 + cc) * (unsigned int)sizeof(uint32_t); //unsigned int pt = readBuffer[1] & 0x7F;
// check if extension // header lenght
if (x) { unsigned int headerlen = (3 + cc) * (unsigned int)sizeof(uint32_t);
// extension header length // check if extension
unsigned int ehl = (((BufferAddr[headerlen + 2] & 0xFF) << 8) | if (x) {
(BufferAddr[headerlen + 3] & 0xFF)); // extension header length
// update header length unsigned int ehl = (((BufferAddr[headerlen + 2] & 0xFF) << 8) |
headerlen += (ehl + 1) * (unsigned int)sizeof(uint32_t); (BufferAddr[headerlen + 3] & 0xFF));
} // update header length
// Check that rtp is version 2 and payload contains multiple of TS packet data headerlen += (ehl + 1) * (unsigned int)sizeof(uint32_t);
if ((v == 2) && (((len - headerlen) % TS_SIZE) == 0) && }
(BufferAddr[headerlen] == TS_SYNC_BYTE)) { // Check that rtp is version 2 and payload contains multiple of TS packet data
// Set argument point to payload in read buffer if ((v == 2) && (((len - headerlen) % TS_SIZE) == 0) &&
memmove(BufferAddr, &BufferAddr[headerlen], (len - headerlen)); (BufferAddr[headerlen] == TS_SYNC_BYTE)) {
return (len - headerlen); // Set argument point to payload in read buffer
} memmove(BufferAddr, &BufferAddr[headerlen], (len - headerlen));
} return (len - headerlen);
} }
} }
} }
} }
}
}
} while (len > 0);
ERROR_IF_RET(len < 0 && errno != EAGAIN, "recvmsg()", return -1);
return 0; return 0;
} }