/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "hadoop/Pipes.hh"
#include "hadoop/SerialUtils.hh"
#include "hadoop/StringUtils.hh"

#include <map>
#include <vector>

#include <errno.h>
#include <netinet/in.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sys/socket.h>
#include <pthread.h>

using std::map;
using std::string;
using std::vector;

using namespace HadoopUtils;

namespace HadoopPipes {

  class JobConfImpl: public JobConf {
  private:
    map<string, string> values;
  public:
    void set(const string& key, const string& value) {
      values[key] = value;
    }

    virtual bool hasKey(const string& key) const {
      return values.find(key) != values.end();
    }

    virtual const string& get(const string& key) const {
      map<string,string>::const_iterator itr = values.find(key);
      if (itr == values.end()) {
        throw Error("Key " + key + " not found in JobConf");
      }
      return itr->second;
    }

    virtual int getInt(const string& key) const {
      const string& val = get(key);
      return toInt(val);
    }

    virtual float getFloat(const string& key) const {
      const string& val = get(key);
      return toFloat(val);
    }

    virtual bool getBoolean(const string&key) const {
      const string& val = get(key);
      return toBool(val);
    }
  };

  class DownwardProtocol {
  public:
    virtual void start(int protocol) = 0;
    virtual void setJobConf(vector<string> values) = 0;
    virtual void setInputTypes(string keyType, string valueType) = 0;
    virtual void runMap(string inputSplit, int numReduces, bool pipedInput)= 0;
    virtual void mapItem(const string& key, const string& value) = 0;
    virtual void runReduce(int reduce, bool pipedOutput) = 0;
    virtual void reduceKey(const string& key) = 0;
    virtual void reduceValue(const string& value) = 0;
    virtual void close() = 0;
    virtual void abort() = 0;
    virtual ~DownwardProtocol() {}
  };

  class UpwardProtocol {
  public:
    virtual void output(const string& key, const string& value) = 0;
    virtual void partitionedOutput(int reduce, const string& key,
                                   const string& value) = 0;
    virtual void status(const string& message) = 0;
    virtual void progress(float progress) = 0;
    virtual void done() = 0;
    virtual void registerCounter(int id, const string& group, 
                                 const string& name) = 0;
    virtual void 
      incrementCounter(const TaskContext::Counter* counter, uint64_t amount) = 0;
    virtual ~UpwardProtocol() {}
  };

  class Protocol {
  public:
    virtual void nextEvent() = 0;
    virtual UpwardProtocol* getUplink() = 0;
    virtual ~Protocol() {}
  };

  class TextUpwardProtocol: public UpwardProtocol {
  private:
    FILE* stream;
    static const char fieldSeparator = '\t';
    static const char lineSeparator = '\n';

    void writeBuffer(const string& buffer) {
      fprintf(stream, quoteString(buffer, "\t\n").c_str());
    }

  public:
    TextUpwardProtocol(FILE* _stream): stream(_stream) {}
    
    virtual void output(const string& key, const string& value) {
      fprintf(stream, "output%c", fieldSeparator);
      writeBuffer(key);
      fprintf(stream, "%c", fieldSeparator);
      writeBuffer(value);
      fprintf(stream, "%c", lineSeparator);
    }

    virtual void partitionedOutput(int reduce, const string& key,
                                   const string& value) {
      fprintf(stream, "parititionedOutput%c%d%c", fieldSeparator, reduce, 
              fieldSeparator);
      writeBuffer(key);
      fprintf(stream, "%c", fieldSeparator);
      writeBuffer(value);
      fprintf(stream, "%c", lineSeparator);
    }

    virtual void status(const string& message) {
      fprintf(stream, "status%c%s%c", fieldSeparator, message.c_str(), 
              lineSeparator);
    }

    virtual void progress(float progress) {
      fprintf(stream, "progress%c%f%c", fieldSeparator, progress, 
              lineSeparator);
    }

    virtual void registerCounter(int id, const string& group, 
                                 const string& name) {
      fprintf(stream, "registerCounter%c%d%c%s%c%s%c", fieldSeparator, id,
              fieldSeparator, group.c_str(), fieldSeparator, name.c_str(), 
              lineSeparator);
    }

    virtual void incrementCounter(const TaskContext::Counter* counter, 
                                  uint64_t amount) {
      fprintf(stream, "incrCounter%c%d%c%ld%c", fieldSeparator, counter->getId(), 
              fieldSeparator, (long)amount, lineSeparator);
    }
    
    virtual void done() {
      fprintf(stream, "done%c", lineSeparator);
    }
  };

  class TextProtocol: public Protocol {
  private:
    FILE* downStream;
    DownwardProtocol* handler;
    UpwardProtocol* uplink;
    string key;
    string value;

    int readUpto(string& buffer, const char* limit) {
      int ch;
      buffer.clear();
      while ((ch = getc(downStream)) != -1) {
        if (strchr(limit, ch) != NULL) {
          return ch;
        }
        buffer += ch;
      }
      return -1;
    }

    static const char* delim;
  public:

    TextProtocol(FILE* down, DownwardProtocol* _handler, FILE* up) {
      downStream = down;
      uplink = new TextUpwardProtocol(up);
      handler = _handler;
    }

    UpwardProtocol* getUplink() {
      return uplink;
    }

    virtual void nextEvent() {
      string command;
      string arg;
      int sep;
      sep = readUpto(command, delim);
      if (command == "mapItem") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(key, delim);
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(value, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->mapItem(key, value);
      } else if (command == "reduceValue") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(value, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->reduceValue(value);
      } else if (command == "reduceKey") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(key, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->reduceKey(key);
      } else if (command == "start") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(arg, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->start(toInt(arg));
      } else if (command == "setJobConf") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(arg, delim);
        int len = toInt(arg);
        vector<string> values(len);
        for(int i=0; i < len; ++i) {
          HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
          sep = readUpto(arg, delim);
          values.push_back(arg);
        }
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->setJobConf(values);
      } else if (command == "setInputTypes") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(key, delim);
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(value, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->setInputTypes(key, value);
      } else if (command == "runMap") {
        string split;
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(split, delim);
        string reduces;
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(reduces, delim);
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(arg, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->runMap(split, toInt(reduces), toBool(arg));
      } else if (command == "runReduce") {
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        sep = readUpto(arg, delim);
        HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
        string piped;
        sep = readUpto(piped, delim);
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->runReduce(toInt(arg), toBool(piped));
      } else if (command == "abort") { 
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->abort();
      } else if (command == "close") {
        HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
        handler->close();
      } else {
        throw Error("Illegal text protocol command " + command);
      }
    }

    ~TextProtocol() {
      delete uplink;
    }
  };
  const char* TextProtocol::delim = "\t\n";

  enum MESSAGE_TYPE {START_MESSAGE, SET_JOB_CONF, SET_INPUT_TYPES, RUN_MAP, 
                     MAP_ITEM, RUN_REDUCE, REDUCE_KEY, REDUCE_VALUE, 
                     CLOSE, ABORT, 
                     OUTPUT=50, PARTITIONED_OUTPUT, STATUS, PROGRESS, DONE,
                     REGISTER_COUNTER, INCREMENT_COUNTER};

  class BinaryUpwardProtocol: public UpwardProtocol {
  private:
    FileOutStream* stream;
  public:
    BinaryUpwardProtocol(FILE* _stream) {
      stream = new FileOutStream();
      HADOOP_ASSERT(stream->open(_stream), "problem opening stream");
    }

    virtual void output(const string& key, const string& value) {
      serializeInt(OUTPUT, *stream);
      serializeString(key, *stream);
      serializeString(value, *stream);
    }

    virtual void partitionedOutput(int reduce, const string& key,
                                   const string& value) {
      serializeInt(PARTITIONED_OUTPUT, *stream);
      serializeInt(reduce, *stream);
      serializeString(key, *stream);
      serializeString(value, *stream);
    }

    virtual void status(const string& message) {
      serializeInt(STATUS, *stream);
      serializeString(message, *stream);
    }

    virtual void progress(float progress) {
      serializeInt(PROGRESS, *stream);
      serializeFloat(progress, *stream);
      stream->flush();
    }

    virtual void done() {
      serializeInt(DONE, *stream);
    }

    virtual void registerCounter(int id, const string& group, 
                                 const string& name) {
      serializeInt(REGISTER_COUNTER, *stream);
      serializeInt(id, *stream);
      serializeString(group, *stream);
      serializeString(name, *stream);
    }

    virtual void incrementCounter(const TaskContext::Counter* counter, 
                                  uint64_t amount) {
      serializeInt(INCREMENT_COUNTER, *stream);
      serializeInt(counter->getId(), *stream);
      serializeLong(amount, *stream);
    }
    
    ~BinaryUpwardProtocol() {
      delete stream;
    }
  };

  class BinaryProtocol: public Protocol {
  private:
    FileInStream* downStream;
    DownwardProtocol* handler;
    BinaryUpwardProtocol * uplink;
    string key;
    string value;

  public:
    BinaryProtocol(FILE* down, DownwardProtocol* _handler, FILE* up) {
      downStream = new FileInStream();
      downStream->open(down);
      uplink = new BinaryUpwardProtocol(up);
      handler = _handler;
    }

    UpwardProtocol* getUplink() {
      return uplink;
    }

    virtual void nextEvent() {
      int32_t cmd;
      cmd = deserializeInt(*downStream);
      switch (cmd) {
      case START_MESSAGE: {
        int32_t prot;
        prot = deserializeInt(*downStream);
        handler->start(prot);
        break;
      }
      case SET_JOB_CONF: {
        int32_t entries;
        entries = deserializeInt(*downStream);
        vector<string> result(entries);
        for(int i=0; i < entries; ++i) {
          string item;
          deserializeString(item, *downStream);
          result.push_back(item);
        }
        handler->setJobConf(result);
        break;
      }
      case SET_INPUT_TYPES: {
        string keyType;
        string valueType;
        deserializeString(keyType, *downStream);
        deserializeString(valueType, *downStream);
        handler->setInputTypes(keyType, valueType);
        break;
      }
      case RUN_MAP: {
        string split;
        int32_t numReduces;
        int32_t piped;
        deserializeString(split, *downStream);
        numReduces = deserializeInt(*downStream);
        piped = deserializeInt(*downStream);
        handler->runMap(split, numReduces, piped);
        break;
      }
      case MAP_ITEM: {
        deserializeString(key, *downStream);
        deserializeString(value, *downStream);
        handler->mapItem(key, value);
        break;
      }
      case RUN_REDUCE: {
        int32_t reduce;
        int32_t piped;
        reduce = deserializeInt(*downStream);
        piped = deserializeInt(*downStream);
        handler->runReduce(reduce, piped);
        break;
      }
      case REDUCE_KEY: {
        deserializeString(key, *downStream);
        handler->reduceKey(key);
        break;
      }
      case REDUCE_VALUE: {
        deserializeString(value, *downStream);
        handler->reduceValue(value);
        break;
      }
      case CLOSE:
        handler->close();
        break;
      case ABORT:
        handler->abort();
        break;
      default:
        HADOOP_ASSERT(false, "Unknown binary command " + toString(cmd));
      }
    }

    virtual ~BinaryProtocol() {
      delete downStream;
      delete uplink;
    }
  };

  /**
   * Define a context object to give to combiners that will let them
   * go through the values and emit their results correctly.
   */
  class CombineContext: public ReduceContext {
  private:
    ReduceContext* baseContext;
    Partitioner* partitioner;
    int numReduces;
    UpwardProtocol* uplink;
    bool firstKey;
    bool firstValue;
    map<string, vector<string> >::iterator keyItr;
    map<string, vector<string> >::iterator endKeyItr;
    vector<string>::iterator valueItr;
    vector<string>::iterator endValueItr;

  public:
    CombineContext(ReduceContext* _baseContext,
                   Partitioner* _partitioner,
                   int _numReduces,
                   UpwardProtocol* _uplink,
                   map<string, vector<string> >& data) {
      baseContext = _baseContext;
      partitioner = _partitioner;
      numReduces = _numReduces;
      uplink = _uplink;
      keyItr = data.begin();
      endKeyItr = data.end();
      firstKey = true;
      firstValue = true;
    }

    virtual const JobConf* getJobConf() {
      return baseContext->getJobConf();
    }

    virtual const std::string& getInputKey() {
      return keyItr->first;
    }

    virtual const std::string& getInputValue() {
      return *valueItr;
    }

    virtual void emit(const std::string& key, const std::string& value) {
      if (partitioner != NULL) {
        uplink->partitionedOutput(partitioner->partition(key, numReduces),
                                  key, value);
      } else {
        uplink->output(key, value);
      }
    }

    virtual void progress() {
      baseContext->progress();
    }

    virtual void setStatus(const std::string& status) {
      baseContext->setStatus(status);
    }

    bool nextKey() {
      if (firstKey) {
        firstKey = false;
      } else {
        ++keyItr;
      }
      if (keyItr != endKeyItr) {
        valueItr = keyItr->second.begin();
        endValueItr = keyItr->second.end();
        firstValue = true;
        return true;
      }
      return false;
    }

    virtual bool nextValue() {
      if (firstValue) {
        firstValue = false;
      } else {
        ++valueItr;
      }
      return valueItr != endValueItr;
    }
    
    virtual Counter* getCounter(const std::string& group, 
                               const std::string& name) {
      return baseContext->getCounter(group, name);
    }

    virtual void incrementCounter(const Counter* counter, uint64_t amount) {
      baseContext->incrementCounter(counter, amount);
    }
  };

  /**
   * A RecordWriter that will take the map outputs, buffer them up and then
   * combine then when the buffer is full.
   */
  class CombineRunner: public RecordWriter {
  private:
    map<string, vector<string> > data;
    int64_t spillSize;
    int64_t numBytes;
    ReduceContext* baseContext;
    Partitioner* partitioner;
    int numReduces;
    UpwardProtocol* uplink;
    Reducer* combiner;
  public:
    CombineRunner(int64_t _spillSize, ReduceContext* _baseContext, 
                  Reducer* _combiner, UpwardProtocol* _uplink, 
                  Partitioner* _partitioner, int _numReduces) {
      numBytes = 0;
      spillSize = _spillSize;
      baseContext = _baseContext;
      partitioner = _partitioner;
      numReduces = _numReduces;
      uplink = _uplink;
      combiner = _combiner;
    }

    virtual void emit(const std::string& key,
                      const std::string& value) {
      numBytes += key.length() + value.length();
      data[key].push_back(value);
      if (numBytes >= spillSize) {
        spillAll();
      }
    }

    virtual void close() {
      spillAll();
    }

  private:
    void spillAll() {
      CombineContext context(baseContext, partitioner, numReduces, 
                             uplink, data);
      while (context.nextKey()) {
        combiner->reduce(context);
      }
      data.clear();
      numBytes = 0;
    }
  };

  class TaskContextImpl: public MapContext, public ReduceContext, 
                         public DownwardProtocol {
  private:
    bool done;
    JobConf* jobConf;
    string key;
    const string* newKey;
    const string* value;
    bool hasTask;
    bool isNewKey;
    bool isNewValue;
    string* inputKeyClass;
    string* inputValueClass;
    string status;
    float progressFloat;
    uint64_t lastProgress;
    bool statusSet;
    Protocol* protocol;
    UpwardProtocol *uplink;
    string* inputSplit;
    RecordReader* reader;
    Mapper* mapper;
    Reducer* reducer;
    RecordWriter* writer;
    Partitioner* partitioner;
    int numReduces;
    const Factory* factory;
    pthread_mutex_t mutexDone;
    std::vector<int> registeredCounterIds;

  public:

    TaskContextImpl(const Factory& _factory) {
      statusSet = false;
      done = false;
      newKey = NULL;
      factory = &_factory;
      jobConf = NULL;
      inputKeyClass = NULL;
      inputValueClass = NULL;
      inputSplit = NULL;
      mapper = NULL;
      reducer = NULL;
      reader = NULL;
      writer = NULL;
      partitioner = NULL;
      protocol = NULL;
      isNewKey = false;
      isNewValue = false;
      lastProgress = 0;
      progressFloat = 0.0f;
      hasTask = false;
      pthread_mutex_init(&mutexDone, NULL);
    }

    void setProtocol(Protocol* _protocol, UpwardProtocol* _uplink) {

      protocol = _protocol;
      uplink = _uplink;
    }

    virtual void start(int protocol) {
      if (protocol != 0) {
        throw Error("Protocol version " + toString(protocol) + 
                    " not supported");
      }
    }

    virtual void setJobConf(vector<string> values) {
      int len = values.size();
      JobConfImpl* result = new JobConfImpl();
      HADOOP_ASSERT(len % 2 == 0, "Odd length of job conf values");
      for(int i=0; i < len; i += 2) {
        result->set(values[i], values[i+1]);
      }
      jobConf = result;
    }

    virtual void setInputTypes(string keyType, string valueType) {
      inputKeyClass = new string(keyType);
      inputValueClass = new string(valueType);
    }

    virtual void runMap(string _inputSplit, int _numReduces, bool pipedInput) {
      inputSplit = new string(_inputSplit);
      reader = factory->createRecordReader(*this);
      HADOOP_ASSERT((reader == NULL) == pipedInput,
                    pipedInput ? "RecordReader defined when not needed.":
                    "RecordReader not defined");
      if (reader != NULL) {
        value = new string();
      }
      mapper = factory->createMapper(*this);
      numReduces = _numReduces;
      if (numReduces != 0) { 
        reducer = factory->createCombiner(*this);
        partitioner = factory->createPartitioner(*this);
      }
      if (reducer != NULL) {
        int64_t spillSize = 100;
        if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
          spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
        }
        writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
                                   uplink, partitioner, numReduces);
      }
      hasTask = true;
    }

    virtual void mapItem(const string& _key, const string& _value) {
      newKey = &_key;
      value = &_value;
      isNewKey = true;
    }

    virtual void runReduce(int reduce, bool pipedOutput) {
      reducer = factory->createReducer(*this);
      writer = factory->createRecordWriter(*this);
      HADOOP_ASSERT((writer == NULL) == pipedOutput,
                    pipedOutput ? "RecordWriter defined when not needed.":
                    "RecordWriter not defined");
      hasTask = true;
    }

    virtual void reduceKey(const string& _key) {
      isNewKey = true;
      newKey = &_key;
    }

    virtual void reduceValue(const string& _value) {
      isNewValue = true;
      value = &_value;
    }
    
    virtual bool isDone() {
      pthread_mutex_lock(&mutexDone);
      bool doneCopy = done;
      pthread_mutex_unlock(&mutexDone);
      return doneCopy;
    }

    virtual void close() {
      pthread_mutex_lock(&mutexDone);
      done = true;
      pthread_mutex_unlock(&mutexDone);
    }

    virtual void abort() {
      throw Error("Aborted by driver");
    }

    void waitForTask() {
      while (!done && !hasTask) {
        protocol->nextEvent();
      }
    }

    bool nextKey() {
      if (reader == NULL) {
        while (!isNewKey) {
          nextValue();
          if (done) {
            return false;
          }
        }
        key = *newKey;
      } else {
        if (!reader->next(key, const_cast<string&>(*value))) {
          pthread_mutex_lock(&mutexDone);
          done = true;
          pthread_mutex_unlock(&mutexDone);
          return false;
        }
        progressFloat = reader->getProgress();
      }
      isNewKey = false;
      if (mapper != NULL) {
        mapper->map(*this);
      } else {
        reducer->reduce(*this);
      }
      return true;
    }

    /**
     * Advance to the next value.
     */
    virtual bool nextValue() {
      if (isNewKey || done) {
        return false;
      }
      isNewValue = false;
      progress();
      protocol->nextEvent();
      return isNewValue;
    }

    /**
     * Get the JobConf for the current task.
     */
    virtual JobConf* getJobConf() {
      return jobConf;
    }

    /**
     * Get the current key. 
     * @return the current key or NULL if called before the first map or reduce
     */
    virtual const string& getInputKey() {
      return key;
    }

    /**
     * Get the current value. 
     * @return the current value or NULL if called before the first map or 
     *    reduce
     */
    virtual const string& getInputValue() {
      return *value;
    }

    /**
     * Mark your task as having made progress without changing the status 
     * message.
     */
    virtual void progress() {
      if (uplink != 0) {
        uint64_t now = getCurrentMillis();
        if (now - lastProgress > 1000) {
          lastProgress = now;
          if (statusSet) {
            uplink->status(status);
            statusSet = false;
          }
          uplink->progress(progressFloat);
        }
      }
    }

    /**
     * Set the status message and call progress.
     */
    virtual void setStatus(const string& status) {
      this->status = status;
      statusSet = true;
      progress();
    }

    /**
     * Get the name of the key class of the input to this task.
     */
    virtual const string& getInputKeyClass() {
      return *inputKeyClass;
    }

    /**
     * Get the name of the value class of the input to this task.
     */
    virtual const string& getInputValueClass() {
      return *inputValueClass;
    }

    /**
     * Access the InputSplit of the mapper.
     */
    virtual const std::string& getInputSplit() {
      return *inputSplit;
    }

    virtual void emit(const string& key, const string& value) {
      progress();
      if (writer != NULL) {
        writer->emit(key, value);
      } else if (partitioner != NULL) {
        int part = partitioner->partition(key, numReduces);
        uplink->partitionedOutput(part, key, value);
      } else {
        uplink->output(key, value);
      }
    }

    /**
     * Register a counter with the given group and name.
     */
    virtual Counter* getCounter(const std::string& group, 
                               const std::string& name) {
      int id = registeredCounterIds.size();
      registeredCounterIds.push_back(id);
      uplink->registerCounter(id, group, name);
      return new Counter(id);
    }

    /**
     * Increment the value of the counter with the given amount.
     */
    virtual void incrementCounter(const Counter* counter, uint64_t amount) {
      uplink->incrementCounter(counter, amount); 
    }

    void closeAll() {
      if (reader) {
        reader->close();
      }
      if (mapper) {
        mapper->close();
      }
      if (reducer) {
        reducer->close();
      }
      if (writer) {
        writer->close();
      }
    }

    virtual ~TaskContextImpl() {
      delete jobConf;
      delete inputKeyClass;
      delete inputValueClass;
      delete inputSplit;
      if (reader) {
        delete value;
      }
      delete reader;
      delete mapper;
      delete reducer;
      delete writer;
      delete partitioner;
      pthread_mutex_destroy(&mutexDone);
    }
  };

  /**
   * Ping the parent every 5 seconds to know if it is alive 
   */
  void* ping(void* ptr) {
    TaskContextImpl* context = (TaskContextImpl*) ptr;
    char* portStr = getenv("mapreduce.pipes.command.port");
    int MAX_RETRIES = 3;
    int remaining_retries = MAX_RETRIES;
    while (!context->isDone()) {
      try{
        sleep(5);
        int sock = -1;
        if (portStr) {
          sock = socket(PF_INET, SOCK_STREAM, 0);
          HADOOP_ASSERT(sock != - 1,
                        string("problem creating socket: ") + strerror(errno));
          sockaddr_in addr;
          addr.sin_family = AF_INET;
          addr.sin_port = htons(toInt(portStr));
          addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
          HADOOP_ASSERT(connect(sock, (sockaddr*) &addr, sizeof(addr)) == 0,
                        string("problem connecting command socket: ") +
                        strerror(errno));

        }
        if (sock != -1) {
          int result = shutdown(sock, SHUT_RDWR);
          HADOOP_ASSERT(result == 0, "problem shutting socket");
          result = close(sock);
          HADOOP_ASSERT(result == 0, "problem closing socket");
        }
        remaining_retries = MAX_RETRIES;
      } catch (Error& err) {
        if (!context->isDone()) {
          fprintf(stderr, "Hadoop Pipes Exception: in ping %s\n", 
                err.getMessage().c_str());
          remaining_retries -= 1;
          if (remaining_retries == 0) {
            exit(1);
          }
        } else {
          return NULL;
        }
      }
    }
    return NULL;
  }

  /**
   * Run the assigned task in the framework.
   * The user's main function should set the various functions using the 
   * set* functions above and then call this.
   * @return true, if the task succeeded.
   */
  bool runTask(const Factory& factory) {
    try {
      TaskContextImpl* context = new TaskContextImpl(factory);
      Protocol* connection;
      char* portStr = getenv("mapreduce.pipes.command.port");
      int sock = -1;
      FILE* stream = NULL;
      FILE* outStream = NULL;
      char *bufin = NULL;
      char *bufout = NULL;
      if (portStr) {
        sock = socket(PF_INET, SOCK_STREAM, 0);
        HADOOP_ASSERT(sock != - 1,
                      string("problem creating socket: ") + strerror(errno));
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(toInt(portStr));
        addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        HADOOP_ASSERT(connect(sock, (sockaddr*) &addr, sizeof(addr)) == 0,
                      string("problem connecting command socket: ") +
                      strerror(errno));

        stream = fdopen(sock, "r");
        outStream = fdopen(sock, "w");

        // increase buffer size
        int bufsize = 128*1024;
        int setbuf;
        bufin = new char[bufsize];
        bufout = new char[bufsize];
        setbuf = setvbuf(stream, bufin, _IOFBF, bufsize);
        HADOOP_ASSERT(setbuf == 0, string("problem with setvbuf for inStream: ")
                                     + strerror(errno));
        setbuf = setvbuf(outStream, bufout, _IOFBF, bufsize);
        HADOOP_ASSERT(setbuf == 0, string("problem with setvbuf for outStream: ")
                                     + strerror(errno));

        connection = new BinaryProtocol(stream, context, outStream);
      } else if (getenv("mapreduce.pipes.commandfile")) {
        char* filename = getenv("mapreduce.pipes.commandfile");
        string outFilename = filename;
        outFilename += ".out";
        stream = fopen(filename, "r");
        outStream = fopen(outFilename.c_str(), "w");
        connection = new BinaryProtocol(stream, context, outStream);
      } else {
        connection = new TextProtocol(stdin, context, stdout);
      }
      context->setProtocol(connection, connection->getUplink());
      pthread_t pingThread;
      pthread_create(&pingThread, NULL, ping, (void*)(context));
      context->waitForTask();
      while (!context->isDone()) {
        context->nextKey();
      }
      context->closeAll();
      connection->getUplink()->done();
      pthread_join(pingThread,NULL);
      delete context;
      delete connection;
      if (stream != NULL) {
        fflush(stream);
      }
      if (outStream != NULL) {
        fflush(outStream);
      }
      fflush(stdout);
      if (sock != -1) {
        int result = shutdown(sock, SHUT_RDWR);
        HADOOP_ASSERT(result == 0, "problem shutting socket");
        result = close(sock);
        HADOOP_ASSERT(result == 0, "problem closing socket");
      }
      if (stream != NULL) {
        //fclose(stream);
      }
      if (outStream != NULL) {
        //fclose(outStream);
      } 
      delete bufin;
      delete bufout;
      return true;
    } catch (Error& err) {
      fprintf(stderr, "Hadoop Pipes Exception: %s\n", 
              err.getMessage().c_str());
      return false;
    }
  }
}