CSL  5.2
RemoteStream.cpp
Go to the documentation of this file.
1 //
2 // RemoteStream.cpp -- a frame stream that sends RFS requests to its "server" to get buffers.
3 // See the copyright notice and acknowledgment of authors in the file COPYRIGHT
4 //
5 
6 #include "RemoteStream.h"
7 
8 #ifndef CSL_WINDOWS
9 #include <sys/param.h>
10 #endif
11 
12 using namespace csl;
13 
14 #ifdef RS_TIMING
15 static struct timeval then, now; // used for getting the real time
16 static long timeVals, thisSec, timeSum; // for printing run-time statistics
17 #endif
18 
19 // RemoteStream Constructor: socket-code heavy lifting
20 // Open up a remote frame stream using a socket to the given remote "server" ip/host
21 
22 RemoteStream::RemoteStream(char * server_name, unsigned short server_port, unsigned chans, unsigned bufSize) {
23  unsigned waitTime = (unsigned) (1000000.0f * (float) bufSize / (float) CGestalt::sample_rate());
24  mNumChannels = chans;
25  mBufferSize = bufSize;
26  mCurrentBuffer = 0;
27  mCurrentFrame = 0;
28 #ifdef CSL_WINDOWS
29  WSADATA localWSA;
30  if (WSAStartup(MAKEWORD(1,1),&localWSA)!= 0)
31  perror("Couldn't do WSAStartup");
32 #endif
33  // set up the packet storage buffers
34  for (int i = 0; i < 2; i++) {
35  mIoBuffers[i] = (sample *) malloc(RS_RESPONSE_PACKET_SIZE);
36  }
37  // call the setup methods
38  if (initSockets(serverName, serverPort))
39  goto error;
40  initPacket();
41  if (connectToServer())
42  goto error;
43  // Now set up the background server semaphore and thread
44  mServerSemaphore = (sem_t *) sem_open("RFS_read_sem", O_CREAT, 0644, 0);
45  if (mServerSemaphore == (sem_t *) SEM_FAILED)
46  perror("RFS: sem_open");
47  // Create the reader thread
48  CSL_CreateThread(RS_read_loop, (void *) this);
49  // Now try to fill the input input buffers
50  for (unsigned i = 0; i < 2; i++) {
51  sem_post(mServerSemaphore); // Trigger the server reader thread
52  sleep_usec(waitTime); // sleep
53  mCurrentBuffer = 1 - mCurrentBuffer;
54  }
55  return;
56 error:
57  logMsg(kLogError, "RemoteStream socket open error");
58  if (mSocket)
59  if (closesocket(mSocket) < 0)
60  perror("Close mSocket");
61  mSocket = -1;
62 }
63 
64 // Now for the socket creation
65 
66 int RemoteStream::initSockets(char * serverName, unsigned short serverPort) {
67  struct hostent * he;
68  // Set up the background thread's semaphore
69  mServerSemaphore = (sem_t *) sem_open("RFS_read_sem", O_CREAT, 0644, 0);
70  if (mServerSemaphore == (sem_t *) SEM_FAILED)
71  perror("RemoteStream: sem_open");
72 
73 #ifdef CSL_DEBUG
74  logMsg("Open RemoteStream output socket"); // Open request socket
75 #endif
76 
77  if ((mSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
78  perror("RemoteStream: socket");
79  return 1;
80  }
81  if ((he = gethostbyname(serverName)) == NULL) { // get the server's IP host info
82  perror("\tRemoteStream: gethostbyname");
83  return 1;
84  }
85  memset((char *) &mServerAddr, 0, sizeof(mServerAddr)); // Set up server address structure
86  mServerAddr.sin_family = AF_INET;
87  mServerAddr.sin_port = htons(serverPort);
88  mServerAddr.sin_addr = *((struct in_addr *) he->h_addr);
89  logMsg("RemoteStream open output socket to %s:%d (= %d) (sz = %d)",
90  inet_ntoa(mServerAddr.sin_addr), serverPort, mSocket, mBufferSize);
91  return 0;
92 }
93 
94 // Set up request packet header
95 
96 void RemoteStream::initPacket() {
97  unsigned char * bString;
98  // Write the structure into the buffer
99  mHeader.magic = RS_PACKET_MAGIC | CSL_CMD_SET_CLIENT;
100  mHeader.frames = mBufferSize;
101  mHeader.channels = mNumChannels;
102  for (int i = 0; i < 2; i++) {
103  bString = (unsigned char *) mIoBuffers[i];
104  memcpy(bString, &mHeader, RS_PACKET_SIZE); // Copy the header to the packets
105  }
106 }
107 
108 // Now send out the "introduction" packet
109 
110 int RemoteStream::connectToServer() {
111  CSL_RS_MSG * packetHeader;
112  ssize_t xferCnt;
113  unsigned command;
114  // connect the socket to the server
115  if (connect(mSocket, (struct sockaddr *) & mServerAddr, sizeof(mServerAddr)) < 0) {
116  perror("\tRemoteStream connect");
117  return 1;
118  } // write the intro packet
119  xferCnt = write(mSocket, (char *) mIoBuffers[0], RS_PACKET_SIZE);
120  if (xferCnt != RS_PACKET_SIZE) {
121  perror("\tRemoteStream intro packet write");
122  return 1;
123  } // read the response
124  xferCnt = read(mSocket, (char *) mIoBuffers[0], RS_PACKET_SIZE);
125  if (xferCnt != RS_PACKET_SIZE) {
126  logMsg(kLogError, "RemoteStream: intro packet read (%d)", xfer_cnt);
127  perror("RemoteStream connect");
128  return 1;
129  } // check the response packet
130  packetHeader = (CSL_RS_MSG *) mIoBuffers[0]; // Cast buffer into a packet header
131  command = packetHeader->magic;
132  if ((command & 0xffffff00) != RS_PACKET_MAGIC) { // Check header fields
133  logMsg(kLogError, "RemoteStream: bad request magic number: %x\n", command);
134  return 1;
135  }
136  logMsg("RemoteStream received server confirmation");
137  return 0;
138 }
139 
140 // Destructor
141 
142 RemoteStream::~RemoteStream() {
143  free(mIoBuffers[0]);
144  free(mIoBuffers[1]);
145  if (mSocket)
146  if (closesocket(mSocket) < 0)
147  perror("RemoteStream: close _sock");
148 }
149 
150 //// nextBuffer method -- copy data and signal the semaphore when you need more ////
151 
152 void RemoteStream::nextBuffer(Buffer &outputBuffer) throw(CException) {
153  unsigned char * out, * buf;
154  unsigned frames = outputBuffer.mNumFrames;
155  unsigned toCopy; // # of bytes to copy per channel
156  unsigned offset;
157  bool split; // whether we're splitting output buffers
158 
159  if ((mCurrentFrame + frames) <= mBufferSize) {
160  split = false;
161  toCopy = frames * sizeof(sample);
162  } else { // else we have to read from both buffers
163  split = true;
164  toCopy = (mBufferSize - mCurrentFrame) * sizeof(sample);
165  }
166  // Copy the output buffer samples
167  for (unsigned i = 0; i < outputBuffer.mNumChannels; i++) {
168  out = (unsigned char *) outputBuffer.buffer(i];
169  offset = RS_PACKET_SIZE + (mCurrentFrame * sizeof(sample))
170  + (i * mBufferSize * sizeof(sample));
171  buf = ((unsigned char *) (mIoBuffers[mCurrentBuffer])) + offset;
172  memcpy(out, buf, toCopy);
173  }
174  mCurrentFrame += frames;
175  if (mCurrentFrame >= mBufferSize) { // If we're done with this buffer
176  mCurrentBuffer = 1 - mCurrentBuffer;
177  mCurrentFrame = 0;
178  sem_post(mServerSemaphore); // Trigger the server reader thread!
179  }
180  if (split) { // if we have to read the start of the other buffer
181  toCopy = (frames * sizeof(sample)) - toCopy;
182  for (unsigned i = 0; i < outputBuffer.mNumChannels; i++) {
183  out = (unsigned char *) outputBuffer.buffer(i];
184  offset = RS_PACKET_SIZE + (i * mBufferSize * sizeof(sample));
185  buf = ((unsigned char *) (mIoBuffers[mCurrentBuffer])) + offset;
186  memcpy(out, buf, toCopy);
187  }
188  mCurrentFrame = toCopy / sizeof(sample);
189  }
190  return;
191 }
192 
193 // Here's the reader thread function; it waits on a semaphore to send a packet to the server
194 
195 extern "C" void * RS_Read_Loop(void * inst) {
196  RemoteStream * rfs = (RemoteStream *) inst;
197  ssize_t bytes_xfer;
198  unsigned command, packet_size;
199  unsigned _numChannels = rfs->channels();
200  unsigned which, _bufferSize;
201  unsigned char * b_str;
202  CSL_RFS_MSG * pkt_header;
203  int sock = rfs->get_sock();
204 #ifdef RS_TIMING
205  thisSec = timeVals = timeSum = 0L;
206 #endif
207 
208  logMsg("\tRemoteStream starting read_loop");
209  while(1) { // -- ENDLESS LOOP --
210  sem_wait(rfs->get_semaphore()); // Wait on the semaphore sent from the reader thread
211  // Set up request packet header
212  which = rfs->get_buffer_switch();
213  _bufferSize = rfs->get_buffer_size();
214  b_str = (unsigned char *) ((rfs->get_io_buffers())[1 - which]);
215  pkt_header = (CSL_RFS_MSG *) b_str; // Cast buffer into a packet header
216  pkt_header->magic = RFS_PACKET_MAGIC | CSL_CMD_NEXT_BUFFER;
217  pkt_header->frames = _bufferSize;
218  pkt_header->channels = _numChannels;
219 #ifdef RS_TIMING
220  GET_TIME(&then);
221 #endif // Send a request packet
222  if (write(sock, (char *) b_str, RFS_PACKET_SIZE) != (int) RFS_PACKET_SIZE) {
223  logMsg(kLogError, "RemoteStream read_loop write");
224  perror("\t\t");
225  continue;
226  } // Read a response packet
227  packet_size = RFS_RESPONSE_PACKET_SIZE;
228  bytes_xfer = read(sock, (char *) b_str, packet_size);
229  if (bytes_xfer != (int) packet_size) {
230  logMsg(kLogError, "RemoteStream read_loop read: %d", bytes_xfer);
231  perror("\t\t");
232  continue;
233  }
234 #ifdef RS_TIMING
235  GET_TIME(&now);
236  rfs->print_time_statistics(&now, &then, &thisSec, &timeSum, &timeVals);
237 #endif // Now process the response buffer
238  command = pkt_header->magic; // Check header/footer fields, ignore packet on error
239  if ((command & 0xffffff00) != RFS_PACKET_MAGIC) {
240  logMsg(kLogError, "RemoteStream: bad response magic number: %x", command);
241  continue;
242  }
243  memcpy(& command, & b_str[packet_size - 4], 4);
244  if (command != RFS_PACKET_MAGIC) { // Check trailing magic number
245  logMsg(kLogError, "RemoteStream: bad response footer: %x", command);
246  continue;
247  }
248  }
249 }
250 
251 // Private thread management function
252 
253 extern "C" int CSL_CreateThread(THREAD_START_ROUTINE pfnThreadProc, void * pvParam) {
254  pthread_t tid;
255  pthread_attr_t attr;
256  int ret;
257 
258  pthread_attr_init(&attr);
259  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
260  pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
261 
262  ret = pthread_create(&tid, &attr, pfnThreadProc, pvParam);
263  return (ret == 0);
264 }