Changeset 356

Timestamp:

04/16/05 18:54:21 (8 years ago)

Author:

ahu

Message:

add joinpipe to Makefile
implement stretches in splitpipe

Location:

trunk/splitpipe

Files:

• Makefile (modified) (2 diffs)
• splitpipe.cc (modified) (13 diffs)

trunk/splitpipe/Makefile

@@ -3 +3 @@
 OUTPUTDIR=splitpipe-$(VERSION)
 
-all: splitpipe
+all: splitpipe joinpipe
 
 clean:
         rm -rf *~ *.o splitpipe 
 
-splitpipe_OBJECTS=splitpipe.o
+common_OBJECTS=misc.o
+splitpipe_OBJECTS=splitpipe.o $(common_OBJECTS)
+joinpipe_OBJECTS=joinpipe.o $(common_OBJECTS)
 
 splitpipe: $(splitpipe_OBJECTS)
         g++ $(splitpipe_OBJECTS) -o $@
+
+joinpipe: $(joinpipe_OBJECTS)
+        g++ $(joinpipe_OBJECTS) -o $@
+
 
 check: 
@@ -23 +29 @@
         tar cvzf $(OUTPUTDIR).tar.gz $(OUTPUTDIR)/
         rm -rf $(OUTPUTDIR)
-        
-        

trunk/splitpipe/splitpipe.cc

@@ -21 +21 @@
 #include <fcntl.h>
 #include <sys/select.h>
+#include <netinet/in.h>
 #include <time.h>
 #include <sys/time.h>
@@ -26 +27 @@
 #include <stdint.h>
 #include <signal.h>
-
+#include "misc.hh"
 #include "ringbuffer.hh"
 
@@ -53 +54 @@
 uint64_t getSize(const char* desc) 
 {
-
   for(struct predef* p=predefinedSizes; p->name; ++p) {
     if(!strcasecmp(p->name, desc))
@@ -69 +69 @@
 }
 
-void unixDie(const string& during)
-{
-  throw runtime_error("during "+string(during)+": "+strerror(errno));
-}
-
-
-void setNonBlocking(int fd)
-{
-  int flags=fcntl(fd,F_GETFL,0);
-  if(flags<0 || fcntl(fd, F_SETFL,flags|O_NONBLOCK) <0)
-    unixDie("Setting filedescriptor to nonblocking failed");
-}
-
-double getTime()
-{
-  struct timeval tv;
-  gettimeofday(&tv, 0);
-  return tv.tv_sec + tv.tv_usec/1000000.0;
-}
-
+pid_t g_pid;
+void waitForOutputCommandToDie()
+{
+  int status;
+  if(waitpid(g_pid, &status, 0) < 0)
+    unixDie("wait on child process");
+  
+  if(WIFEXITED(status))
+    cerr<<"\nsplitpipe: output command exited with status "<<WEXITSTATUS(status)<<endl;
+  else {
+    cerr<<"\nsplitpipe: output command exited abnormally";
+    if(WIFSIGNALED(status))
+      cerr<<", by signal "<<WTERMSIG(status);
+    cerr<<endl;
+  }
+}
+
+
+void outputGaveEof(int outputfd)
+{
+  cerr<<"\nsplitpipe: output command gave EOF, waiting for it to exit"<<endl;
+  close(outputfd);
+  waitForOutputCommandToDie();
+  cerr<<"\nsplitpipe: future versions of splitpipe may allow you to continue, but for now.. exit\n";
+  exit(EXIT_FAILURE);
+}
 
 void usage()
@@ -150 +156 @@
 }
 
-pid_t g_pid;
+
 
 int spawnOutputThread()
@@ -196 +202 @@
 }
 
-void waitForOutputCommandToDie()
-{
-  int status;
-  if(waitpid(g_pid, &status, 0) < 0)
-    unixDie("wait on child process");
-  
-  if(WIFEXITED(status))
-    cerr<<"\nsplitpipe: output command exited with status "<<WEXITSTATUS(status)<<endl;
-  else {
-    cerr<<"\nsplitpipe: output command exited abnormally";
-    if(WIFSIGNALED(status))
-      cerr<<", by signal "<<WTERMSIG(status);
-    cerr<<endl;
-  }
-}
 
 int main(int argc, char** argv)
@@ -260 +251 @@
   int outputfd;
   uint64_t amountOutput=0;
+  uint16_t leftInStretch=0;
+  int numStretches=0;
 
   char *buffer = new char[parameters.bufferSize];
@@ -266 +259 @@
     cerr<<"Prebuffering before starting output script..";
 
-  bool d_firstchunk=true;
+  bool d_firstchunk=true; // first chunk does not get the 'press enter' stuff
 
   while(1) {
@@ -278 +271 @@
         cerr<<"splitpipe: reload media, if necessary, and press enter to continue"<<endl;
         waitForUser();
-
       }
 
@@ -332 +324 @@
         rb.get(&rbuffer, &lenAvailable);
 
+        if(!leftInStretch) {
+          leftInStretch=min((size_t)0xffff, lenAvailable);
+
+          if(parameters.debug) 
+            cerr<<"splitpipe: starting a stretch of "<<leftInStretch<<" bytes"<<endl;
+          uint16_t amount=htons(leftInStretch);
+          ret=writen(outputfd, &amount, sizeof(amount),"write of meta-data to output command");
+          if(!ret)
+            outputGaveEof(outputfd);
+          
+          numStretches++;
+          break; // go past select again, not sure if this is needed
+        }
+
         uint64_t leftInChunk=parameters.chunkSize - amountOutput;
 
         size_t len=min((uint64_t)lenAvailable, leftInChunk);
         
+        len=min(len, (size_t)leftInStretch);
+
         if(!len) {
           cerr<<"\nsplitpipe: output a full chunk, waiting for output command to exit..\n";
@@ -357 +365 @@
 
         if(!ret) {
-          cerr<<"\nsplitpipe: output command gave EOF, waiting for it to exit"<<endl;
-          close(outputfd);
-          waitForOutputCommandToDie();
-          cerr<<"\nsplitpipe: future versions of splitpipe may allow you to continue, but for now.. exit\n";
-          exit(EXIT_FAILURE);
+          outputGaveEof(outputfd);
         }
         if(parameters.debug) 
@@ -367 +371 @@
         rb.advance(ret);
 
-        amountOutput+=ret;
-
-        totalBytesOutput+=ret;
+        amountOutput += ret;
+        leftInStretch -= ret;
+        totalBytesOutput += ret;
 
         if(parameters.debug) 
@@ -386 +390 @@
     waitForOutputCommandToDie();
   }
+  cerr<<"splitpipe: output "<<numStretches<<" stretches\n";
 }
 catch(exception &e)