libcm is a C development framework with an emphasis on audio signal processing applications.
選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

cmUdpPort.c 14KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547
  1. #include "cmPrefix.h"
  2. #include "cmGlobal.h"
  3. #include "cmRpt.h"
  4. #include "cmErr.h"
  5. #include "cmCtx.h"
  6. #include "cmMem.h"
  7. #include "cmMallocDebug.h"
  8. #include "cmThread.h"
  9. #include <sys/socket.h>
  10. #include <netinet/in.h>
  11. #include <arpa/inet.h>
  12. #include <fcntl.h>
  13. #include <unistd.h> // close
  14. #include "cmUdpPort.h"
  15. #define cmUdp_SYS_ERR (-1)
  16. #define cmUdp_NULL_SOCK (-1)
  17. enum
  18. {
  19. kIsConnectedUdpFl = 0x01,
  20. kIsBlockingUdpFl = 0x02,
  21. kQueueingUdpFl = 0x04
  22. };
  23. typedef struct
  24. {
  25. cmErr_t err;
  26. int sockH;
  27. cmUdpCallback_t cbFunc;
  28. void* cbArg;
  29. unsigned timeOutMs;
  30. unsigned flags;
  31. cmThreadH_t thH;
  32. cmTs1p1cH_t qH;
  33. unsigned recvBufByteCnt;
  34. char* tempBuf;
  35. unsigned timeOutCnt;
  36. unsigned recvCnt;
  37. unsigned queCbCnt;
  38. unsigned errCnt;
  39. } cmUdp_t;
  40. cmUdpH_t cmUdpNullHandle = cmSTATIC_NULL_HANDLE;
  41. #define _cmUdpClear_errno() errno = 0
  42. cmUdp_t* _cmUdpHandleToPtr( cmUdpH_t h )
  43. {
  44. cmUdp_t* p = (cmUdp_t*)h.h;
  45. assert(p != NULL);
  46. return p;
  47. }
  48. cmUdpRC_t _cmUdpFinal( cmUdp_t* p )
  49. {
  50. cmUdpRC_t rc = kOkUdpRC;
  51. if( cmThreadIsValid(p->thH) )
  52. if( cmThreadDestroy(&p->thH) != kOkThRC )
  53. return cmErrMsg(&p->err,kThreadFailUdpRC,"Listener thread destroy failed.");
  54. if( cmTs1p1cIsValid(p->qH) )
  55. if( cmTs1p1cDestroy(&p->qH) != kOkThRC )
  56. cmErrMsg(&p->err,kQueueFailUdpRC,"Receive data queue destroy failed.");
  57. cmMemPtrFree(&p->tempBuf);
  58. // close the socket
  59. if( p->sockH != cmUdp_NULL_SOCK )
  60. {
  61. _cmUdpClear_errno();
  62. if( close(p->sockH) != 0 )
  63. cmErrSysMsg(&p->err,kSockCloseFailUdpRC,errno,"The socket close failed." );
  64. p->sockH = cmUdp_NULL_SOCK;
  65. }
  66. return rc;
  67. }
  68. cmUdpRC_t _cmUdpFree( cmUdp_t* p )
  69. {
  70. cmUdpRC_t rc;
  71. if((rc = _cmUdpFinal(p)) != kOkUdpRC )
  72. return rc;
  73. cmMemFree(p);
  74. return rc;
  75. }
  76. cmUdpRC_t _cmUdpInitAddr( cmUdp_t* p, const char* addrStr, cmUdpPort_t portNumber, struct sockaddr_in* retAddrPtr )
  77. {
  78. memset(retAddrPtr,0,sizeof(struct sockaddr_in));
  79. if( addrStr == NULL )
  80. retAddrPtr->sin_addr.s_addr = htonl(INADDR_ANY);
  81. else
  82. {
  83. _cmUdpClear_errno();
  84. if(inet_pton(AF_INET,addrStr,&retAddrPtr->sin_addr) == 0 )
  85. //if(( retAddrPtr->sin_addr.s_addr = inet_addr(addrStr)) == INADDR_NONE )
  86. return cmErrSysMsg(&p->err,kPtoNFailUdpRC,errno, "The network address string '%s' could not be converted to a netword address structure.",cmStringNullGuard(addrStr) );
  87. }
  88. //retAddrPtr->sin_len = sizeof(struct sockaddr_in);
  89. retAddrPtr->sin_family = AF_INET;
  90. retAddrPtr->sin_port = htons(portNumber);
  91. return kOkUdpRC;
  92. }
  93. cmUdpRC_t _cmUdpConnect( cmUdp_t* p, const char* remoteAddr, cmUdpPort_t remotePort )
  94. {
  95. struct sockaddr_in addr;
  96. cmUdpRC_t rc;
  97. // create the remote address
  98. if((rc = _cmUdpInitAddr(p, remoteAddr, remotePort, &addr )) != kOkUdpRC )
  99. return rc;
  100. _cmUdpClear_errno();
  101. // ... and connect this socket to the remote address/port
  102. if( connect(p->sockH, (struct sockaddr*)&addr, sizeof(addr)) == cmUdp_SYS_ERR )
  103. return cmErrSysMsg(&p->err,kSockConnectFailUdpRC, errno, "Socket connect failed." );
  104. p->flags = cmSetFlag(p->flags,kIsConnectedUdpFl);
  105. return rc;
  106. }
  107. cmUdpRC_t cmUdpAlloc( cmCtx_t* ctx, cmUdpH_t* hp )
  108. {
  109. cmUdpRC_t rc;
  110. if((rc = cmUdpFree(hp)) != kOkUdpRC )
  111. return rc;
  112. cmUdp_t* p = cmMemAllocZ(cmUdp_t,1);
  113. cmErrSetup(&p->err,&ctx->rpt,"UDP Port");
  114. p->sockH = cmUdp_NULL_SOCK;
  115. hp->h = p;
  116. return rc;
  117. }
  118. cmUdpRC_t cmUdpFree( cmUdpH_t* hp )
  119. {
  120. cmUdpRC_t rc = kOkUdpRC;
  121. if( hp == NULL || cmUdpIsValid(*hp)==false)
  122. return rc;
  123. cmUdp_t* p = _cmUdpHandleToPtr(*hp);
  124. if((rc = _cmUdpFree(p)) != kOkUdpRC )
  125. return rc;
  126. hp->h = NULL;
  127. return rc;
  128. }
  129. cmUdpRC_t cmUdpInit(
  130. cmUdpH_t h,
  131. cmUdpPort_t port,
  132. unsigned flags,
  133. cmUdpCallback_t cbFunc,
  134. void* cbArg,
  135. const char* remoteAddr,
  136. cmUdpPort_t remotePort,
  137. unsigned recvBufByteCnt,
  138. unsigned timeOutMs )
  139. {
  140. cmUdpRC_t rc;
  141. struct sockaddr_in addr;
  142. cmUdp_t* p = _cmUdpHandleToPtr(h);
  143. if((rc = _cmUdpFinal(p)) != kOkUdpRC )
  144. return rc;
  145. _cmUdpClear_errno();
  146. // get a handle to the socket
  147. if(( p->sockH = socket( AF_INET, SOCK_DGRAM, IPPROTO_UDP ) ) == cmUdp_SYS_ERR )
  148. return cmErrSysMsg(&p->err, kSockCreateFailUdpRC, errno, "Socket create failed." );
  149. // create the local address
  150. if((rc = _cmUdpInitAddr(p, NULL, port, &addr )) != kOkUdpRC )
  151. goto errLabel;
  152. // bind the socket to a local address/port
  153. if( (bind( p->sockH, (struct sockaddr*)&addr, sizeof(addr))) == cmUdp_SYS_ERR )
  154. {
  155. rc = cmErrSysMsg(&p->err,kSockBindFailUdpRC,errno,"Socket bind failed." );
  156. goto errLabel;
  157. }
  158. // if a remote addr was given connect this socket to it
  159. if( remoteAddr != NULL )
  160. if((rc = _cmUdpConnect(p,remoteAddr,remotePort)) != kOkUdpRC )
  161. goto errLabel;
  162. // if this socket should block
  163. if( cmIsFlag(flags,kBlockingUdpFl) )
  164. {
  165. struct timeval timeOut;
  166. // set the socket time out
  167. timeOut.tv_sec = timeOutMs/1000;
  168. timeOut.tv_usec = (timeOutMs - (timeOut.tv_sec * 1000)) * 1000;
  169. if( setsockopt( p->sockH, SOL_SOCKET, SO_RCVTIMEO, &timeOut, sizeof(timeOut) ) == cmUdp_SYS_ERR )
  170. {
  171. rc = cmErrSysMsg(&p->err,kSockOptSetFailUdpRC,errno, "Attempt to set the socket timeout failed." );
  172. goto errLabel;
  173. }
  174. p->flags = cmSetFlag(p->flags,kIsBlockingUdpFl);
  175. }
  176. else
  177. {
  178. int opts;
  179. // get the socket options flags
  180. if( (opts = fcntl(p->sockH,F_GETFL)) < 0 )
  181. {
  182. rc = cmErrSysMsg(&p->err,kSockOptSetFailUdpRC,errno, "Attempt to get the socket options flags failed." );
  183. goto errLabel;
  184. }
  185. opts = (opts | O_NONBLOCK);
  186. // set the socket options flags
  187. if(fcntl(p->sockH,F_SETFL,opts) < 0)
  188. {
  189. rc = cmErrSysMsg(&p->err,kSockOptSetFailUdpRC,errno, "Attempt to set the socket to non-blocking failed." );
  190. goto errLabel;
  191. }
  192. }
  193. if( recvBufByteCnt != 0 )
  194. p->tempBuf = cmMemAlloc(char,recvBufByteCnt );
  195. p->timeOutMs = timeOutMs;
  196. p->cbFunc = cbFunc;
  197. p->cbArg = cbArg;
  198. p->recvBufByteCnt = recvBufByteCnt;
  199. p->timeOutCnt = 0;
  200. p->recvCnt = 0;
  201. p->queCbCnt = 0;
  202. p->errCnt = 0;
  203. if( cmIsFlag(flags,kNoQueueUdpFl) == false )
  204. p->flags = cmSetFlag(p->flags,kQueueingUdpFl);
  205. errLabel:
  206. if( rc != kOkUdpRC )
  207. _cmUdpFree(p);
  208. return rc;
  209. }
  210. cmUdpRC_t cmUdpFinal( cmUdpH_t h )
  211. {
  212. cmUdp_t* p = _cmUdpHandleToPtr(h);
  213. return _cmUdpFinal(p);
  214. }
  215. bool cmUdpIsValid( cmUdpH_t h )
  216. { return h.h != NULL; }
  217. cmUdpRC_t cmUdpConnect( cmUdpH_t h, const char* remoteAddr, cmUdpPort_t remotePort )
  218. {
  219. cmUdp_t* p = _cmUdpHandleToPtr(h);
  220. return _cmUdpConnect(p,remoteAddr,remotePort);
  221. }
  222. cmUdpRC_t cmUdpSend( cmUdpH_t h, const char* data, unsigned dataByteCnt )
  223. {
  224. cmUdp_t* p = _cmUdpHandleToPtr(h);
  225. _cmUdpClear_errno();
  226. if( cmIsFlag(p->flags,kIsConnectedUdpFl) == false )
  227. return cmErrMsg(&p->err,kNotConnectedUdpRC,"cmUdpSend() only works with connected sockets.");
  228. if( send( p->sockH, data, dataByteCnt, 0 ) == cmUdp_SYS_ERR )
  229. return cmErrSysMsg(&p->err,kSockSendFailUdpRC,errno,"Send failed.");
  230. return kOkUdpRC;
  231. }
  232. cmUdpRC_t cmUdpSendTo( cmUdpH_t h, const char* data, unsigned dataByteCnt, const struct sockaddr_in* remoteAddr )
  233. {
  234. cmUdp_t* p = _cmUdpHandleToPtr(h);
  235. _cmUdpClear_errno();
  236. if( sendto(p->sockH, data, dataByteCnt, 0, (struct sockaddr*)remoteAddr, sizeof(*remoteAddr)) == cmUdp_SYS_ERR )
  237. return cmErrSysMsg(&p->err,kSockSendFailUdpRC,errno,"SendTo failed.");
  238. return kOkUdpRC;
  239. }
  240. cmUdpRC_t cmUdpSend2( cmUdpH_t h, const char* data, unsigned dataByteCnt, const char* remoteAddr, cmUdpPort_t remotePort )
  241. {
  242. cmUdpRC_t rc;
  243. cmUdp_t* p = _cmUdpHandleToPtr(h);
  244. struct sockaddr_in addr;
  245. if((rc = _cmUdpInitAddr(p,remoteAddr,remotePort,&addr)) != kOkUdpRC )
  246. return rc;
  247. return cmUdpSendTo( h, data, dataByteCnt, &addr );
  248. }
  249. cmUdpRC_t cmUdpRecv( cmUdpH_t h, char* data, unsigned dataByteCnt, struct sockaddr_in* fromAddr, unsigned* recvByteCntPtr )
  250. {
  251. cmUdp_t* p = _cmUdpHandleToPtr(h);
  252. cmUdpRC_t rc = kOkUdpRC;
  253. ssize_t retVal = 0;
  254. socklen_t sizeOfRemoteAddr = fromAddr==NULL ? 0 : sizeof(struct sockaddr_in);
  255. _cmUdpClear_errno();
  256. if( recvByteCntPtr != NULL )
  257. *recvByteCntPtr = 0;
  258. if((retVal = recvfrom(p->sockH, data, dataByteCnt, 0, (struct sockaddr*)fromAddr, &sizeOfRemoteAddr )) == cmUdp_SYS_ERR )
  259. return cmErrSysMsg(&p->err,kSockRecvFailUdpRC,errno,"recvFrom() failed.");
  260. if( recvByteCntPtr != NULL )
  261. *recvByteCntPtr = retVal;
  262. return rc;
  263. }
  264. bool _cmUdpThreadCb(void* param)
  265. {
  266. cmUdp_t* p = (cmUdp_t*)param;
  267. fd_set rdSet;
  268. struct timeval timeOut;
  269. // setup the select() call
  270. FD_ZERO(&rdSet);
  271. FD_SET(p->sockH, &rdSet );
  272. timeOut.tv_sec = p->timeOutMs/1000;
  273. timeOut.tv_usec = (p->timeOutMs - (timeOut.tv_sec * 1000)) * 1000;
  274. // NOTE; select() takes the highest socket value plus one of all the sockets in all the sets.
  275. switch( select(p->sockH+1,&rdSet,NULL,NULL,&timeOut) )
  276. {
  277. case -1: // error
  278. if( errno != EINTR )
  279. cmErrSysMsg(&p->err,kSockSelectFailUdpRC,errno,"Select failed.");
  280. ++p->errCnt;
  281. break;
  282. case 0: // select() timed out
  283. ++p->timeOutCnt;
  284. break;
  285. case 1: // (> 0) count of ready descripters
  286. if( FD_ISSET(p->sockH,&rdSet) )
  287. {
  288. struct sockaddr_in remoteAddr;
  289. socklen_t addrByteCnt = sizeof(remoteAddr);
  290. ssize_t retByteCnt;
  291. _cmUdpClear_errno();
  292. ++p->recvCnt;
  293. // recv the incoming msg into p->tempBuf[]
  294. if(( retByteCnt = recvfrom( p->sockH, p->tempBuf, p->recvBufByteCnt, 0, (struct sockaddr*)&remoteAddr, &addrByteCnt )) == cmUdp_SYS_ERR )
  295. cmErrSysMsg(&p->err,kSockRecvFailUdpRC,errno,"recvfrom() failed.");
  296. else
  297. {
  298. // check for overflow
  299. if( retByteCnt == p->recvBufByteCnt )
  300. cmErrMsg(&p->err,kRecvBufOverflowUdpRC,"The receive buffer requires more than %i bytes.",p->recvBufByteCnt);
  301. else
  302. {
  303. // if queueing is enabled
  304. if( cmIsFlag(p->flags,kQueueingUdpFl ) )
  305. {
  306. // enqueue the msg - with the source address appended after the data
  307. const void* msgPtrArray[] = { p->tempBuf, &remoteAddr, p->tempBuf };
  308. unsigned msgByteCntArray[] = { retByteCnt, sizeof(remoteAddr) };
  309. if( cmTs1p1cEnqueueSegMsg( p->qH, msgPtrArray, msgByteCntArray, 2 ) != kOkThRC )
  310. cmErrMsg(&p->err,kQueueFailUdpRC,"A received msg containing %i bytes was not queued.",retByteCnt);
  311. }
  312. else // if queueing is not enabled - transmit the data directly via the callback
  313. if( p->cbFunc != NULL )
  314. {
  315. p->cbFunc(p->cbArg,p->tempBuf,retByteCnt,&remoteAddr);
  316. }
  317. }
  318. }
  319. }
  320. break;
  321. default:
  322. { assert(0); }
  323. } // switch
  324. return true;
  325. }
  326. cmRC_t _cmUdpQueueCb(void* userCbPtr, unsigned msgByteCnt, const void* msgDataPtr )
  327. {
  328. cmUdp_t* p = (cmUdp_t*)userCbPtr;
  329. if( p->cbFunc != NULL )
  330. {
  331. struct sockaddr_in addr;
  332. assert( msgByteCnt >= sizeof(addr));
  333. const char* dataPtr = (const char*)msgDataPtr;
  334. // the address of the data source is apppended to the data bytes.
  335. const char* addrPtr = dataPtr + msgByteCnt - sizeof(addr);
  336. memcpy(&addr,addrPtr,sizeof(addr));
  337. // make the receive callback
  338. p->cbFunc(p->cbArg,dataPtr,msgByteCnt-sizeof(addr),&addr);
  339. ++p->queCbCnt;
  340. }
  341. return cmOkRC;
  342. }
  343. cmUdpRC_t cmUdpEnableListen( cmUdpH_t h, bool enableFl )
  344. {
  345. cmUdp_t* p = _cmUdpHandleToPtr(h);
  346. if( cmThreadIsValid(p->thH) == false && enableFl == true)
  347. {
  348. if(cmThreadCreate(&p->thH,_cmUdpThreadCb,p,p->err.rpt) != kOkThRC )
  349. return cmErrMsg(&p->err,kThreadFailUdpRC,"Listener thread create failed.");
  350. if(cmTs1p1cCreate(&p->qH,p->recvBufByteCnt,_cmUdpQueueCb,p,p->err.rpt) != kOkThRC )
  351. return cmErrMsg(&p->err,kQueueFailUdpRC,"Listener data queue create failed.");
  352. }
  353. if( cmThreadIsValid(p->thH) )
  354. if( cmThreadPause( p->thH, enableFl ? 0 : kPauseThFl ) != kOkThRC )
  355. return cmErrMsg(&p->err,kThreadFailUdpRC,"The listener thread failed to %s.", enableFl ? "pause" : "un-pause" );
  356. return kOkUdpRC;
  357. }
  358. bool cmUdpIsQueueEnabled( cmUdpH_t h )
  359. {
  360. cmUdp_t* p = _cmUdpHandleToPtr(h);
  361. return cmIsFlag(p->flags,kQueueingUdpFl);
  362. }
  363. void cmUdpQueueEnable( cmUdpH_t h, bool enableFl )
  364. {
  365. cmUdp_t* p = _cmUdpHandleToPtr(h);
  366. p->flags = cmSetFlag(p->flags,kQueueingUdpFl);
  367. }
  368. unsigned cmUdpAvailDataByteCount( cmUdpH_t h )
  369. {
  370. cmUdp_t* p = _cmUdpHandleToPtr(h);
  371. return cmTs1p1cIsValid(p->qH) ? cmTs1p1cDequeueMsgByteCount( p->qH ) : 0;
  372. }
  373. cmUdpRC_t cmUdpGetAvailData( cmUdpH_t h, char* data, unsigned* dataByteCntPtr, struct sockaddr_in* fromAddr )
  374. {
  375. cmUdp_t* p = _cmUdpHandleToPtr(h);
  376. unsigned availByteCnt;
  377. // if a received msg is queued
  378. if( (availByteCnt = cmTs1p1cAvailByteCount(p->qH)) > 0 )
  379. {
  380. // all msg's must have at least a source address
  381. assert( availByteCnt >= sizeof(*fromAddr) );
  382. // get the size of the return buffer (or 0 if there is no return buffer)
  383. unsigned dataByteCnt = (data != NULL && dataByteCntPtr != NULL) ? *dataByteCntPtr : 0;
  384. if( dataByteCnt == 0 )
  385. data = NULL;
  386. // dequeue the msg - if data==NULL then the data will be returned by
  387. // a call to the callback function provided in cmUdpAlloc().
  388. if( cmTs1p1cDequeueMsg(p->qH, data, dataByteCnt ) != kOkThRC )
  389. return cmErrMsg(&p->err,kQueueFailUdpRC,"Data dequeue failed.");
  390. // if a return buffer was given
  391. if( data != NULL )
  392. {
  393. assert( dataByteCntPtr != NULL );
  394. // the source address is appended to the end of the data
  395. const char* addrPtr = data + availByteCnt - sizeof(*fromAddr);
  396. // copy out the source address
  397. if( fromAddr != NULL )
  398. memcpy(fromAddr,addrPtr,sizeof(*fromAddr));
  399. // subtract the address size from the total msg size
  400. *dataByteCntPtr = availByteCnt - sizeof(*fromAddr);
  401. }
  402. }
  403. return kOkUdpRC;
  404. }
  405. void cmUdpReport( cmUdpH_t h, cmRpt_t* rpt )
  406. {
  407. cmUdp_t* p = _cmUdpHandleToPtr(h);
  408. cmRptPrintf(rpt,"time-out:%i recv:%i queue cb:%i\n",p->timeOutCnt,p->recvCnt,p->queCbCnt);
  409. }
  410. cmUdpRC_t cmUdpInitAddr( cmUdpH_t h, const char* addrStr, cmUdpPort_t portNumber, struct sockaddr_in* retAddrPtr )
  411. {
  412. cmUdp_t* p = _cmUdpHandleToPtr(h);
  413. return _cmUdpInitAddr(p,addrStr,portNumber,retAddrPtr);
  414. }