tools/tesh2/src/writer.c

   1 #include <writer.h>
   2 #include <command.h>
   3
   4 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(tesh);
   5
   6 static void*
   7 writer_start_routine(void* p);
   8
   9 writer_t
  10 writer_new(command_t command)
  11 {
  12         writer_t writer = xbt_new0(s_writer_t, 1);
  13
  14         writer->thread = NULL;
  15         writer->command = command;
  16         writer->written = xbt_os_sem_init(0);
  17         writer->can_write = xbt_os_sem_init(0);
  18
  19         writer->done = 0;
  20
  21         return writer;
  22 }
  23
  24 void
  25 writer_free(writer_t* writer)
  26 {
  27
  28         /*xbt_os_sem_destroy((*writer)->started);
  29         xbt_os_sem_destroy((*writer)->can_write);*/
  30
  31         free(*writer);
  32         *writer = NULL;
  33 }
  34
  35 void
  36 writer_write(writer_t writer)
  37 {
  38         writer->thread = xbt_os_thread_create("", writer_start_routine, writer);
  39 }
  40
  41 #ifdef WIN32
  42 static void*
  43 writer_start_routine(void* p)
  44 {
  45         writer_t writer = (writer_t)p;
  46         command_t command = writer->command;
  47
  48         char* input = (char*)(command->context->input->data);
  49
  50         DWORD number_of_bytes_to_write = command->context->input->used;
  51         DWORD number_of_bytes_written = 0;
  52
  53
  54         while(!command->failed && !command->interrupted && !command->successeded && ! writer->failed && ! writer->broken_pipe && number_of_bytes_to_write)
  55         {
  56                 if(!WriteFile(writer->command->stdin_fd, input, number_of_bytes_to_write, &number_of_bytes_written, NULL))
  57                 {
  58                         if(GetLastError() ==  ERROR_NO_DATA)
  59                                 writer->broken_pipe = 1;
  60                         else
  61                         writer->failed = 1;
  62
  63                 }
  64                 else
  65                 {
  66                         input += number_of_bytes_written;
  67                         number_of_bytes_to_write -= number_of_bytes_written;
  68                 }
  69         }
  70
  71         command->context->input->data[0]='\0';
  72         command->context->input->used=0;
  73
  74         if(writer->failed  && !command->successeded && !command->failed && !command->interrupted)
  75         {
  76                 command_kill(command);
  77                 exit_code = EWRITE;
  78                 command_handle_failure(command, csr_write_failure);
  79         }
  80         /*else if(writer->broken_pipe && !command->successeded && !command->failed && !command->interrupted)
  81         {
  82                 command_kill(command);
  83                 command_handle_failure(command, csr_write_pipe_broken);
  84         }*/
  85
  86
  87         return NULL;
  88 }
  89 #else
  90 /*static void*
  91 writer_start_routine(void* p)
  92 {
  93         writer_t writer = (writer_t)p;
  94         command_t command = writer->command;
  95         int number_of_bytes_written = 0;
  96         int number_of_bytes_to_write = command->context->input->used;
  97         char* input = (char*)(command->context->input->data);
  98         int got;
  99
 100         xbt_os_sem_release(writer->started);
 101
 102
 103
 104         while(!command->failed && !command->interrupted && !command->successeded && ! writer->failed && ! writer->broken_pipe && (number_of_bytes_written < number_of_bytes_to_write))
 105         {
 106                 got = write(writer->command->stdin_fd, input + number_of_bytes_written, number_of_bytes_to_write - number_of_bytes_written);
 107
 108                 if(got > 0)
 109                         number_of_bytes_written += got;
 110
 111                 if(got < 0)
 112                 {
 113                         if(errno == EPIPE)
 114                         {
 115                                 writer->broken_pipe = 1;
 116                         }
 117                         else if(errno != EINTR && errno != EAGAIN && errno != EPIPE)
 118                         {
 119                                 writer->failed = 1;
 120                         }
 121                 }
 122
 123                 if(got <= 0)
 124                         usleep(100);
 125         }
 126
 127         command->context->input->data[0]='\0';
 128         command->context->input->used=0;
 129
 130         if(writer->failed  && !command->successeded && !command->failed && !command->interrupted)
 131         {
 132                 command_kill(command);
 133                 exit_code = EWRITE;
 134                 command_handle_failure(command, csr_write_failure);
 135         }
 136         else if(writer->broken_pipe && !command->successeded && !command->failed && !command->interrupted)
 137         {
 138                 command_kill(command);
 139                 exit_code = EWRITEPIPE;
 140                 command_handle_failure(command, csr_write_pipe_broken);
 141         }
 142
 143
 144         close(command->stdin_fd);
 145         command->stdin_fd = INDEFINITE_FD;
 146
 147         return NULL;
 148 }*/
 149
 150 static void*
 151 writer_start_routine(void* p)
 152 {
 153         writer_t writer = (writer_t)p;
 154         command_t command = writer->command;
 155         int number_of_bytes_to_write = command->context->input->used;
 156         char* input = (char*)(command->context->input->data);
 157         int got;
 158         int released = 0;
 159
 160
 161         xbt_os_sem_acquire(writer->can_write);
 162
 163
 164         while(!command->failed && !command->interrupted && !command->successeded && number_of_bytes_to_write > 0)
 165         {
 166                 got = number_of_bytes_to_write > PIPE_BUF ? PIPE_BUF : number_of_bytes_to_write;
 167                 got = write(writer->command->stdin_fd, input, got );
 168
 169                 if(got < 0)
 170                 {
 171                         if(EINTR == errno)
 172                                 continue;
 173
 174                         else if(EAGAIN == errno)
 175                         {/* the pipe is full */
 176                                 if(!released)
 177                                 {
 178                                         xbt_os_sem_release(writer->written);
 179                                         released = 1;
 180                                 }
 181
 182                                 continue;
 183                         }
 184                         else if(EPIPE == errno)
 185                         {
 186                                 writer->broken_pipe = 1;
 187                                 break;
 188                         }
 189                         else
 190                         {
 191                                 writer->failed = 1;
 192                                 break;
 193                         }
 194
 195                 }
 196
 197                 number_of_bytes_to_write -= got;
 198                 input += got;
 199
 200                 if(got == 0)
 201                         xbt_os_thread_yield();
 202
 203         }
 204
 205         if(!released)
 206         {
 207                 xbt_os_sem_release(writer->written);
 208                 released = 1;
 209         }
 210
 211
 212         close(command->stdin_fd);
 213         command->stdin_fd = INDEFINITE_FD;
 214
 215         command->context->input->data[0]='\0';
 216         command->context->input->used=0;
 217
 218         if(writer->failed  && !command->successeded && !command->failed && !command->interrupted)
 219         {
 220                 command_kill(command);
 221                 exit_code = EWRITE;
 222                 command_handle_failure(command, csr_write_failure);
 223         }
 224         else if(writer->broken_pipe && !command->successeded && !command->failed && !command->interrupted)
 225         {
 226                 command_kill(command);
 227                 exit_code = EWRITEPIPE;
 228                 command_handle_failure(command, csr_write_pipe_broken);
 229         }
 230
 231         writer->done = 1;
 232
 233         return NULL;
 234
 235 }
 236
 237 #endif
 238 void
 239 writer_wait(writer_t writer)
 240 {
 241         xbt_os_thread_join(writer->thread, NULL);
 242
 243 }