x10 version of pipe_sample

MyCpp.cpp

#include <iostream>
#include <vector>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <unistd.h>
#include <x10/lang/Rail.h>
#include <x10/lang/String.h>

int popen2(char*const* argv, int *fd_r, int *fd_w) {

  const size_t READ = 0;
  const size_t WRITE = 1;

  int pipe_child2parent[2];
  int pipe_parent2child[2];
  int pid;

  if (pipe(pipe_child2parent) != 0 ||
    pipe(pipe_parent2child) != 0 ||
    (pid = fork()) < 0
    ) {
    perror("popen2");
    close(pipe_child2parent[READ]);
    close(pipe_child2parent[WRITE]);
    close(pipe_parent2child[READ]);
    close(pipe_parent2child[WRITE]);
    return 1;
  }

  if (pid == 0) { // at the child process
    close(pipe_parent2child[WRITE]);
    close(pipe_child2parent[READ]);
    dup2(pipe_parent2child[READ], 0);
    dup2(pipe_child2parent[WRITE], 1);
    close(pipe_parent2child[READ]);
    close(pipe_child2parent[WRITE]);

    // execute subprocess
    if (execvp(argv[0], argv) < 0) {
        perror("popen2");
        close(pipe_parent2child[READ]);
        close(pipe_child2parent[WRITE]);
        return 1;
    }
  }

  // at the parent process
  close(pipe_parent2child[READ]);
  close(pipe_child2parent[WRITE]);

  *fd_r = pipe_child2parent[READ];
  *fd_w = pipe_parent2child[WRITE];

  return pid;
}


void lntrim(char *str) {  
  char *p;  
  p = strchr(str, '\n');  
  if(p != NULL) {  
    *p = '\0';  
  }  
}

x10::lang::Rail<x10::lang::String*>* readLines(FILE* fp_r) {
  size_t buf_size = 512;
  char* buf = (char*) malloc(buf_size);

  std::vector<x10::lang::String*> lines;

  int len = getline( &buf, &buf_size, fp_r );
  lntrim(buf);
  while( strlen(buf) > 0 ) {
    std::cerr << "[INFO] reading: " << buf << std::endl;
    x10::lang::String* val = x10::lang::String::_make(buf, false);
    lines.push_back(val);
    len = getline( &buf, &buf_size, fp_r );
    lntrim(buf);
  }
  std::cerr << "[INFO] reading task end" << std::endl;
  free(buf);

  size_t num_lines = lines.size();
  x10::lang::Rail<x10::lang::String*>* arr(x10::lang::Rail<x10::lang::String*>::_make(num_lines));
  for (int i = 0; i < num_lines; i++) {
    arr->__set(i, lines[i]);
  }
  return arr;
}

void writeLines(FILE* fp_w, x10::lang::Rail<x10::lang::String*>* lines) {
  for( size_t i=0; i < lines->FMGL(size); i++) {
    x10::lang::String* p_s = lines->raw[i];
    std::cerr << "[INFO] writing: " << p_s->c_str() << std::endl;
    fprintf(fp_w, "%s\n", p_s->c_str() );
  }
  fprintf(fp_w, "\n");
  std::cerr << "[INFO] writing end" << std::endl;
  fflush(fp_w);
}

void launchSubProcess( x10::lang::Rail<x10::lang::String*>* x10_argv, long* fps_pid) {

  size_t size = x10_argv->FMGL(size);
  char** argv = new char*[size+1];
  for( size_t i=0; i < x10_argv->FMGL(size); i++) {
    x10::lang::String* ps = x10_argv->raw[i];
    argv[i] = const_cast<char*>( ps->c_str() );
  }
  argv[size] = NULL;

  int fd_r, fd_w;
  int pid = popen2( argv, &fd_r, &fd_w);
  FILE* fp_r = fdopen(fd_r, "r");
  FILE* fp_w = fdopen(fd_w, "w");
  fps_pid[0] = pid;
  fps_pid[1] = (long) fp_r;
  fps_pid[2] = (long) fp_w;

  delete [] argv;
}

MyCpp.hpp

#include <cstdio>
#include <x10/lang/String.h>

void launchSubProcess( x10::lang::Rail<x10::lang::String*>* x10_argv, long* fps_pid);
x10::lang::Rail<x10::lang::String*>* readLines(FILE* fp_r);
void writeLines(FILE* fp_w, x10::lang::Rail<x10::lang::String*>* lines);

pipe_sample.x10

import x10.io.Console;
import x10.compiler.Native;
import x10.compiler.NativeCPPInclude;
import x10.compiler.NativeCPPCompilationUnit;


@NativeCPPInclude("MyCpp.hpp")
@NativeCPPCompilationUnit("MyCpp.cpp")

class Main {

  public static def main( args: Rail[String] ) {
  
    val fpsPid: Rail[Long] = new Rail[Long](3);

    if( args.size == 0 ) {
      Console.ERR.println("[ERR] No argument is given. You must specify the command of searcher");
      throw new Exception("No arguments given.");
    }

    val out: Rail[String];
    launchSubProcess( args, fpsPid );
    out = readLines( fpsPid(1) );
    for( x in out ) { Console.OUT.println(x); }

    for( i in 0..3 ) {
      val results: Rail[String] = ["abc" as String, "def", "ghi"];
      writeLines( fpsPid(2), results ); // as Rail[String] );
      val o: Rail[String] = readLines( fpsPid(1) );
      for( x in o ) { Console.OUT.println(x); }
    }
  }

  @Native("c++", "launchSubProcess( #1, (long*) &((#2)->raw[0]) )")
  native static def launchSubProcess( argv: Rail[String], fp_and_pid: Rail[Long] ): void;

  @Native("c++", "readLines( (FILE*)(#1) )")
  native static def readLines( fp_r: Long ): Rail[String];

  @Native("c++", "writeLines( (FILE*)(#1), #2 )")
  native static def writeLines( fp_w: Long, lines: Rail[String] ): void;

}

test.py

from __future__ import print_function
import sys

def submit(i,cmd,point,seed):
    s_point = " ".join( [str(x) for x in point] )
    out = "%d %s %d %s %d" % (i,cmd,i,s_point,seed)
    print(out)

for i in range(3):
    submit(i,"./foobar.out",(i+1,i+2,3-i), i+1000)
print("")

def readlines_until_empty():
    lines = []
    while True:
        line = sys.stdin.readline().rstrip()
        if not line:
            break
        lines.append(line)
    return lines


while True:
    lines = readlines_until_empty()
    if len(lines) == 0:
        break
    for line in lines:
        submit(1,line.rstrip(),(2,3,4),1234)
    print("")

sys.stderr.write("ending python\n")