// util/kaldi-table-inl.h
  
  // Copyright 2009-2011    Microsoft Corporation
  //                2013    Johns Hopkins University (author: Daniel Povey)
  //                2016    Xiaohui Zhang
  
  // See ../../COPYING for clarification regarding multiple authors
  //
  // Licensed 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
  //
  // THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  // KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED
  // WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
  // MERCHANTABLITY OR NON-INFRINGEMENT.
  // See the Apache 2 License for the specific language governing permissions and
  // limitations under the License.
  
  
  #ifndef KALDI_UTIL_KALDI_TABLE_INL_H_
  #define KALDI_UTIL_KALDI_TABLE_INL_H_
  
  #include <algorithm>
  #include <string>
  #include <thread>
  #include <utility>
  #include <vector>
  #include <errno.h>
  #include "util/kaldi-io.h"
  #include "util/kaldi-holder.h"
  #include "util/text-utils.h"
  #include "util/stl-utils.h"  // for StringHasher.
  #include "util/kaldi-semaphore.h"
  
  
  namespace kaldi {
  
  /// \addtogroup table_impl_types
  /// @{
  
  template<class Holder> class SequentialTableReaderImplBase {
   public:
    typedef typename Holder::T T;
    // note that Open takes rxfilename not rspecifier.  Open will only be
    // called on a just-allocated object.
    virtual bool Open(const std::string &rxfilename) = 0;
    // Done() should be called on a successfully opened, not-closed object.
    // only throws if called at the wrong time (i.e. code error).
    virtual bool Done() const = 0;
    // Returns true if the reader is open [i.e. Open() succeeded and
    // the user has not called Close()]
    virtual bool IsOpen() const = 0;
    // Returns the current key; it is valid to call this if Done() returned false.
    // Only throws on code error (i.e. called at the wrong time).
    virtual std::string Key() = 0;
    // Returns the value associated with the current key.  Valid to call it if
    // Done() returned false.  It throws if the value could not be read.  [However
    // if you use the ,p modifier it will never throw, unless you call it at the
    // wrong time, i.e. unless there is a code error.]
    virtual T &Value() = 0;
    virtual void FreeCurrent() = 0;
    // move to the next object.  This won't throw unless called wrongly (e.g. on
    // non-open archive.]
    virtual void Next() = 0;
    // Close the table.  Returns its status as bool so it won't throw, unless
    // called wrongly [i.e. on non-open archive.]
    virtual bool Close() = 0;
    // SwapHolder() is not part of the public interface of SequentialTableReader.
    // It should be called when it would be valid to call Value() or FreeCurrent()
    // (i.e. when a value is stored), and after this it's not valid to get the
    // value any more until you call Next().  It swaps the contents of
    // this->holder_ with those of 'other_holder'.  It's needed as part of how
    // we implement SequentialTableReaderBackgroundImpl.
    virtual void SwapHolder(Holder *other_holder) = 0;
    SequentialTableReaderImplBase() { }
    virtual ~SequentialTableReaderImplBase() { }  // throws.
   private:
    KALDI_DISALLOW_COPY_AND_ASSIGN(SequentialTableReaderImplBase);
  };
  
  // This is the implementation for SequentialTableReader
  // when it's actually a script file.
  template<class Holder>  class SequentialTableReaderScriptImpl:
        public SequentialTableReaderImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    SequentialTableReaderScriptImpl(): state_(kUninitialized) { }
  
    // You may call Open from states kUninitialized and kError.
    // It may leave the object in any of the states.
    virtual bool Open(const std::string &rspecifier) {
      if (state_ != kUninitialized && state_ != kError)
        if (!Close())  // call Close() yourself to suppress this exception.
          KALDI_ERR << "Error closing previous input: "
                    << "rspecifier was " << rspecifier_;
      bool binary;
      rspecifier_ = rspecifier;
      RspecifierType rs = ClassifyRspecifier(rspecifier, &script_rxfilename_,
                                             &opts_);
      KALDI_ASSERT(rs == kScriptRspecifier);
      if (!script_input_.Open(script_rxfilename_, &binary)) {  // Failure on Open
        KALDI_WARN << "Failed to open script file "
                   << PrintableRxfilename(script_rxfilename_);
        state_ = kUninitialized;
        return false;
      } else {  // Open succeeded.
        if (binary) {
          KALDI_WARN << "Script file should not be binary file.";
          SetErrorState();
          return false;
        } else {
          state_ = kFileStart;
          Next();
          if (state_ == kError)
            return false;
          // any other status, including kEof, is OK from the point of view of
          // the 'open' function (empty scp file is not inherently an error).
          return true;
        }
      }
    }
  
    virtual bool IsOpen() const {
      switch (state_) {
        case kEof: case kHaveScpLine: case kHaveObject: case kHaveRange:
          return true;
        case kUninitialized: case kError:
          return false;
        default: KALDI_ERR << "IsOpen() called on invalid object.";
          // note: kFileStart is not a valid state for the user to call a member
          // function (we never return from a public function in this state).
          return false;
      }
    }
  
    virtual bool Done() const {
      switch (state_) {
        case kHaveScpLine: case kHaveObject: case kHaveRange: return false;
        case kEof: case kError: return true;  // Error condition, like Eof, counts
          // as Done(); the destructor/Close() will inform the user of the error.
        default: KALDI_ERR << "Done() called on TableReader object at the wrong"
            " time.";
          return false;
      }
    }
  
    virtual std::string Key() {
      // Valid to call this whenever Done() returns false.
      switch (state_) {
        case kHaveScpLine: case kHaveObject: case kHaveRange: break;
        default:
          // coding error.
          KALDI_ERR << "Key() called on TableReader object at the wrong time.";
      }
      return key_;
    }
  
    T &Value() {
      if (!EnsureObjectLoaded())
        KALDI_ERR << "Failed to load object from "
                  << PrintableRxfilename(data_rxfilename_)
                  << " (to suppress this error, add the permissive "
                  << "(p, ) option to the rspecifier.";
      // Because EnsureObjectLoaded() returned with success, we know
      // that if range_ is nonempty (i.e. a range was requested), the
      // state will be kHaveRange.
      if (state_ == kHaveRange) {
        return range_holder_.Value();
      } else {
        KALDI_ASSERT(state_ == kHaveObject);
        return holder_.Value();
      }
    }
  
    void FreeCurrent() {
      if (state_ == kHaveObject) {
        holder_.Clear();
        state_ = kHaveScpLine;
      } else if (state_ == kHaveRange) {
        range_holder_.Clear();
        state_ = kHaveObject;
      } else {
        KALDI_WARN << "FreeCurrent called at the wrong time.";
      }
    }
  
    void SwapHolder(Holder *other_holder) {
      // call Value() to ensure we have a value, and ignore its return value while
      // suppressing compiler warnings by casting to void.  It will cause the
      // program to die with KALDI_ERR if we couldn't get a value.
      (void) Value();
      // At this point we know that we successfully loaded an object,
      // and if there was a range specified, it's in range_holder_.
      if (state_ == kHaveObject) {
        holder_.Swap(other_holder);
        state_ = kHaveScpLine;
      } else if (state_ == kHaveRange) {
        range_holder_.Swap(other_holder);
        state_ = kHaveObject;
        // This indicates that we still have the base object (but no range).
      } else {
        KALDI_ERR << "Code error";
      }
      // Note: after this call there may be some junk left in range_holder_ or
      // holder_, but it won't matter.  We avoid calling Clear() on them, as this
      // function needs to be lightweight for the 'bg' feature to work well.
    }
  
    // Next goes to the next object.
    // It can leave the object in most of the statuses, but
    // the only circumstances under which it will return are:
    //  either:
    //  - if Done() returned true, i.e. kError or kEof.
    //  or:
    //  - in non-permissive mode, status kHaveScpLine or kHaveObjecct
    //  - in permissive mode, only when we successfully have an object,
    //    which means either (kHaveObject and range_.empty()), or
    //    kHaveRange.
    void Next() {
      while (1) {
        NextScpLine();
        if (Done()) return;
        if (opts_.permissive) {
          // Permissive mode means, when reading scp files, we treat keys whose
          // scp entry cannot be read as nonexistent.  This means trying to read.
          if (EnsureObjectLoaded()) return;  // Success.
          // else try the next scp line.
        } else {
          return;  // We go the next key; Value() will crash if we can't read the
                   // object on the scp line.
        }
      }
    }
  
    // This function may be entered at in any state.  At exit, the object will be
    // in state kUninitialized.  It only returns false in the situation where we
    // were at the end of the stream (kEof) and the script_input_ was a pipe and
    // it ended with error status; this is so that we can catch errors from
    // programs that we invoked via a pipe.
    virtual bool Close() {
      int32 status = 0;
      if (script_input_.IsOpen())
        status = script_input_.Close();
      if (data_input_.IsOpen())
        data_input_.Close();
      range_holder_.Clear();
      holder_.Clear();
      if (!this->IsOpen())
        KALDI_ERR << "Close() called on input that was not open.";
      StateType old_state = state_;
      state_ = kUninitialized;
      if (old_state == kError || (old_state == kEof && status != 0)) {
        if (opts_.permissive) {
          KALDI_WARN << "Close() called on scp file with read error, ignoring the"
              " error because permissive mode specified.";
          return true;
        } else {
          return false;  // User will do something with the error status.
        }
      } else {
        return true;
      }
      // Possible states                                          Return value
      // kLoadSucceeded/kRangeSucceeded/kRangeFailed              true
      // kError (if opts_.permissive)                             true
      // kError (if !opts_.permissive)                            false
      // kEof (if script_input_.Close() && !opts.permissive)      false
      // kEof (if !script_input_.Close() || opts.permissive)      true
      // kUninitialized/kFileStart/kHaveScpLine                   true
      // kUnitialized                                             true
    }
  
    virtual ~SequentialTableReaderScriptImpl() {
      if (this->IsOpen() && !Close())
        KALDI_ERR << "TableReader: reading script file failed: from scp "
                        << PrintableRxfilename(script_rxfilename_);
    }
   private:
  
    // Function EnsureObjectLoaded() ensures that we have fully loaded any object
    // (including object range) associated with the current key, and returns true
    // on success (i.e. we have the object) and false on failure.
    //
    // Possible entry states: kHaveScpLine, kLoadSucceeded, kRangeSucceeded
    //
    // Possible exit states: kHaveScpLine, kLoadSucceeded, kRangeSucceeded.
    //
    // Note: the return status has information that cannot be deduced from
    // just the exit state.  If the object could not be loaded we go to state
    // kHaveScpLine but return false; and if the range was requested but
    // could not be extracted, we go to state kLoadSucceeded but return false.
    bool EnsureObjectLoaded() {
      if (!(state_ == kHaveScpLine || state_ == kHaveObject ||
            state_ == kHaveRange))
        KALDI_ERR << "Invalid state (code error)";
  
      if (state_ == kHaveScpLine) {  // need to load the object into holder_.
        bool ans;
        // note, NULL means it doesn't read the binary-mode header
        if (Holder::IsReadInBinary()) {
          ans = data_input_.Open(data_rxfilename_, NULL);
        } else {
          ans = data_input_.OpenTextMode(data_rxfilename_);
        }
        if (!ans) {
          KALDI_WARN << "Failed to open file "
                     << PrintableRxfilename(data_rxfilename_);
          return false;
        } else {
          if (holder_.Read(data_input_.Stream())) {
            state_ = kHaveObject;
          } else {  // holder_ will not contain data.
            KALDI_WARN << "Failed to load object from "
                       << PrintableRxfilename(data_rxfilename_);
            return false;
          }
        }
      }
      // OK, at this point the state must be either
      // kHaveObject or kHaveRange.
      if (range_.empty()) {
        // if range_ is the empty string, we should not be in the state
        // kHaveRange.
        KALDI_ASSERT(state_ == kHaveObject);
        return true;
      }
      // range_ is nonempty.
      if (state_ == kHaveRange) {
        // range was already extracted, so there nothing to do.
        return true;
      }
      // OK, range_ is nonempty and state_ is kHaveObject.  We attempt to extract
      // the range object.  Note: ExtractRange() will throw with KALDI_ERR if the
      // object type doesn't support ranges.
      if (!range_holder_.ExtractRange(holder_, range_)) {
        KALDI_WARN  << "Failed to load object from "
                    << PrintableRxfilename(data_rxfilename_)
                    << "[" << range_ << "]";
        return false;
      } else {
        state_ = kHaveRange;
        return true;
      }
    }
  
    void SetErrorState() {
      state_ = kError;
      script_input_.Close();
      data_input_.Close();
      holder_.Clear();
      range_holder_.Clear();
    }
  
    // Reads the next line in the script file.
    // Possible entry states: kHaveObject, kHaveRange, kHaveScpLine, kFileStart.
    // Possible exit states: kEof, kError, kHaveScpLine, kHaveObject.
    void NextScpLine() {
      switch (state_) {  // Check and simplify the state.
        case kHaveRange:
          range_holder_.Clear();
          state_ = kHaveObject;
          break;
        case kHaveScpLine: case kHaveObject: case kFileStart: break;
        default:
          // No other states are valid to call Next() from.
          KALDI_ERR << "Reading script file: Next called wrongly.";
      }
      // at this point the state will be kHaveObject, kHaveScpLine, or kFileStart.
      std::string line;
      if (getline(script_input_.Stream(), line)) {
        // After extracting "key" from "line", we put the rest
        // of "line" into "rest", and then extract data_rxfilename_
        // (e.g. 1.ark:100) and possibly the range_ specifer
        // (e.g. [1:2,2:10]) from "rest".
        std::string data_rxfilename, rest;
        SplitStringOnFirstSpace(line, &key_, &rest);
        if (!key_.empty() && !rest.empty()) {
          // Got a valid line.
          if (rest[rest.size()-1] == ']') {
            if(!ExtractRangeSpecifier(rest, &data_rxfilename, &range_)) {
              KALDI_WARN << "Reading rspecifier '" << rspecifier_
                         << ", cannot make sense of scp line "
                         << line;
              SetErrorState();
              return;
            }
          } else {
            data_rxfilename = rest;
            range_ = "";
          }
          bool filenames_equal = (data_rxfilename_ == data_rxfilename);
          if (!filenames_equal)
            data_rxfilename_ = data_rxfilename;
          if (state_ == kHaveObject) {
            if (!filenames_equal) {
              holder_.Clear();
              state_ = kHaveScpLine;
            }
            // else leave state_ at kHaveObject and leave the object in the
            // holder.
          } else {
            state_ = kHaveScpLine;
          }
        } else {
          KALDI_WARN << "We got an invalid line in the scp file. "
                     << "It should look like: some_key 1.ark:10, got: "
                     << line;
          SetErrorState();
        }
      } else {
        state_ = kEof;  // there is nothing more in the scp file.  Might as well
                        // close input streams as we don't need them.
        script_input_.Close();
        if (data_input_.IsOpen())
          data_input_.Close();
        holder_.Clear();  // clear the holder if it was nonempty.
        range_holder_.Clear();  // clear the range holder if it was nonempty.
      }
    }
  
    std::string rspecifier_;  // the rspecifier that this class was opened with.
    RspecifierOptions opts_;  // options.
    std::string script_rxfilename_;  // rxfilename of the script file.
  
    Input script_input_;  // Input object for the .scp file
    Input data_input_;   // Input object for the entries in the script file;
                         // we make this a class member instead of a local variable,
                         // so that rspecifiers of the form filename:byte-offset,
                         // e.g. foo.ark:12345, can be handled using fseek().
  
    Holder holder_;       // Holds the object.
    Holder range_holder_; // Holds the partial object corresponding to the object
                          // range specifier 'range_'; this is only used when
                          // 'range_' is specified, i.e. when the .scp file
                          // contains lines of the form rspecifier[range], like
                          // foo.ark:242[0:9] (representing a row range of a
                          // matrix).
  
  
    std::string key_;  // the key of the current scp line we're processing
    std::string data_rxfilename_;  // the rxfilename corresponding to the current key
    std::string range_;  // the range of object corresponding to the current key, if an
                         // object range was specified in the script file, else "".
  
    enum StateType {
      //  Summary of the states this object can be in (state_).
      //
      //                (*) Does holder_ contain the object corresponding to
      //                    data_rxfilename_ ?
      //                    (*) Does range_holder_ contain a range object?
      //                         (*) is script_input_ open?
      //                             (*) are key_, data_rxfilename_ and range_ [if applicable] set?
      //
      kUninitialized, // no  no  no  no            Uninitialized or closed object.
      kFileStart,     // no  no  yes no            We just opened the .scp file (we'll never be in this
                      //                           state when a user-visible function is called.)
      kEof,           // no  no  no  no            We did Next() and found eof in script file.
      kError,         // no  no  no  no            Error reading or parsing script file.
      kHaveScpLine,   // no  no  yes yes           Have a line of the script file but nothing else.
      kHaveObject,    // yes no  yes yes           holder_ contains an object but range_holder_ does not.
      kHaveRange,     // yes yes yes yes           we have the range object in range_holder_ (implies
                      //                           range_ nonempty).
    } state_;
  
  
  };
  
  
  // This is the implementation for SequentialTableReader
  // when it's an archive.  Note that the archive format is:
  // key1 [space] object1 key2 [space]
  // object2 ... eof.
  // "object1" is the output of the Holder::Write function and will
  // typically contain a binary header (in binary mode) and then
  // the output of object.Write(os, binary).
  // The archive itself does not care whether it is in binary
  // or text mode, for reading purposes.
  
  template<class Holder>  class SequentialTableReaderArchiveImpl:
        public SequentialTableReaderImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    SequentialTableReaderArchiveImpl(): state_(kUninitialized) { }
  
    virtual bool Open(const std::string &rspecifier) {
      if (state_ != kUninitialized) {
        if (!Close()) {  // call Close() yourself to suppress this exception.
          if (opts_.permissive)
            KALDI_WARN << "Error closing previous input "
                "(only warning, since permissive mode).";
          else
            KALDI_ERR << "Error closing previous input.";
        }
      }
      rspecifier_ = rspecifier;
      RspecifierType rs = ClassifyRspecifier(rspecifier,
                                             &archive_rxfilename_,
                                             &opts_);
      KALDI_ASSERT(rs == kArchiveRspecifier);
  
      bool ans;
      // NULL means don't expect binary-mode header
      if (Holder::IsReadInBinary())
        ans = input_.Open(archive_rxfilename_, NULL);
      else
        ans = input_.OpenTextMode(archive_rxfilename_);
      if (!ans) {  // header.
        KALDI_WARN << "Failed to open stream "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kUninitialized;  // Failure on Open
        return false;  // User should print the error message.
      }
      state_ = kFileStart;
      Next();
      if (state_ == kError) {
        KALDI_WARN << "Error beginning to read archive file (wrong filename?): "
                   << PrintableRxfilename(archive_rxfilename_);
        input_.Close();
        state_ = kUninitialized;
        return false;
      }
      KALDI_ASSERT(state_ == kHaveObject || state_ == kEof);
      return true;
    }
  
    virtual void Next() {
      switch (state_) {
        case kHaveObject:
          holder_.Clear();
          break;
        case kFileStart: case kFreedObject:
          break;
        default:
          KALDI_ERR << "Next() called wrongly.";
      }
      std::istream &is = input_.Stream();
      is.clear();  // Clear any fail bits that may have been set... just in case
      // this happened in the Read function.
      is >> key_;  // This eats up any leading whitespace and gets the string.
      if (is.eof()) {
        state_ = kEof;
        return;
      }
      if (is.fail()) {  // This shouldn't really happen, barring file-system
                        // errors.
        KALDI_WARN << "Error reading archive "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kError;
        return;
      }
      int c;
      if ((c = is.peek()) != ' ' && c != '\t' && c != '
  ') {  // We expect a
                                                       // space ' ' after the key.
        // We also allow tab [which is consumed] and newline [which is not], just
        // so we can read archives generated by scripts that may not be fully
        // aware of how this format works.
        KALDI_WARN << "Invalid archive file format: expected space after key "
                   << key_ << ", got character "
                   << CharToString(static_cast<char>(is.peek())) << ", reading "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kError;
        return;
      }
      if (c != '
  ') is.get();  // Consume the space or tab.
      if (holder_.Read(is)) {
        state_ = kHaveObject;
        return;
      } else {
        KALDI_WARN << "Object read failed, reading archive "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kError;
        return;
      }
    }
  
    virtual bool IsOpen() const {
      switch (state_) {
        case kEof: case kError: case kHaveObject: case kFreedObject: return true;
        case kUninitialized: return false;
        default: KALDI_ERR << "IsOpen() called on invalid object.";  // kFileStart
          // is not valid state for user to call something on.
          return false;
      }
    }
  
    virtual bool Done() const {
      switch (state_) {
        case kHaveObject:
          return false;
        case kEof: case kError:
          return true;  // Error-state counts as Done(), but destructor
          // will fail (unless you check the status with Close()).
        default:
          KALDI_ERR << "Done() called on TableReader object at the wrong time.";
          return false;
      }
    }
  
    virtual std::string Key() {
      // Valid to call this whenever Done() returns false
      switch (state_) {
        case kHaveObject: break;  // only valid case.
        default:
          // coding error.
          KALDI_ERR << "Key() called on TableReader object at the wrong time.";
      }
      return key_;
    }
  
    T &Value() {
      switch (state_) {
        case kHaveObject:
          break;  // only valid case.
        default:
          // coding error.
          KALDI_ERR << "Value() called on TableReader object at the wrong time.";
      }
      return holder_.Value();
    }
  
    virtual void FreeCurrent() {
      if (state_ == kHaveObject) {
        holder_.Clear();
        state_ = kFreedObject;
      } else {
        KALDI_WARN << "FreeCurrent called at the wrong time.";
      }
    }
  
    void SwapHolder(Holder *other_holder) {
      // call Value() to ensure we have a value, and ignore its return value while
      // suppressing compiler warnings by casting to void.
      (void) Value();
      if (state_ == kHaveObject) {
        holder_.Swap(other_holder);
        state_ = kFreedObject;
      } else {
        KALDI_ERR << "SwapHolder called at the wrong time "
                     "(error related to ',bg' modifier).";
      }
    }
  
    virtual bool Close() {
      // To clean up, Close() also closes the Input object if
      // it's open.  It will succeed if the stream was not in an error state,
      // and the Input object isn't in an error state  we've found eof in the archive.
      if (!this->IsOpen())
        KALDI_ERR << "Close() called on TableReader twice or otherwise wrongly.";
      int32 status = 0;
      if (input_.IsOpen())
        status = input_.Close();
      if (state_ == kHaveObject)
        holder_.Clear();
      StateType old_state = state_;
      state_ = kUninitialized;
      if (old_state == kError || (old_state == kEof && status != 0)) {
        if (opts_.permissive) {
          KALDI_WARN << "Error detected closing TableReader for archive "
                     << PrintableRxfilename(archive_rxfilename_)
                     << " but ignoring "
                     << "it as permissive mode specified.";
          return true;
        } else {
          return false;
        }
      } else {
        return true;
      }
    }
  
    virtual ~SequentialTableReaderArchiveImpl() {
      if (this->IsOpen() && !Close())
        KALDI_ERR << "TableReader: error detected closing archive "
                  << PrintableRxfilename(archive_rxfilename_);
    }
   private:
    Input input_;  // Input object for the archive
    Holder holder_;     // Holds the object.
    std::string key_;
    std::string rspecifier_;
    std::string archive_rxfilename_;
    RspecifierOptions opts_;
    enum StateType {  //  [The state of the reading process]        [does holder_ [is input_
      //                                                     have object]   open]
      kUninitialized,  // Uninitialized or closed.                  no         no
      kFileStart,      // [state we use internally: just opened.]   no         yes
      kEof,     // We did Next() and found eof in archive           no         no
      kError,   // Some other error                                 no         no
      kHaveObject,  // We read the key and the object after it.     yes        yes
      kFreedObject,  // The user called FreeCurrent().              no         yes
    } state_;
  };
  
  // this is for when someone adds the 'th' modifier; it wraps around the basic
  // implementation and allows it to do the reading in a background thread.
  template<class Holder>
  class SequentialTableReaderBackgroundImpl:
        public SequentialTableReaderImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    SequentialTableReaderBackgroundImpl(
        SequentialTableReaderImplBase<Holder> *base_reader):
        base_reader_(base_reader) {}
  
    // This function ignores the rxfilename argument.
    // We use the same function signature as the regular Open(),
    // for convenience.
    virtual bool Open(const std::string &rxfilename) {
      KALDI_ASSERT(base_reader_ != NULL &&
                   base_reader_->IsOpen());  // or code error.
      {
        thread_ = std::thread(SequentialTableReaderBackgroundImpl<Holder>::run,
                              this);
      }
  
      if (!base_reader_->Done())
        Next();
      return true;
    }
  
    virtual bool IsOpen() const {
      // Close() sets base_reader_ to NULL, and we never initialize this object
      // with a non-open base_reader_, so no need to check if it's open.
      return base_reader_ != NULL;
    }
  
    void RunInBackground() {
      try {
        // This function is called in the background thread.  The whole point of
        // the background thread is that we don't want to do the actual reading
        // (inside Next()) in the foreground.
        while (base_reader_ != NULL && !base_reader_->Done()) {
          consumer_sem_.Signal();
          // Here is where the consumer process (parent thread) gets to do its
          // stuff.  Principally it calls SwapHolder()-- a shallow swap that is
          // cheap.
          producer_sem_.Wait();
          // we check that base_reader_ is not NULL in case Close() was
          // called in the main thread.
          if (base_reader_ != NULL)
            base_reader_->Next();   //  here is where the work happens.
        }
        // this signal will be waited on in the Next() function of the foreground
        // thread if it is still running, or Close() otherwise.
        consumer_sem_.Signal();
        // this signal may be waited on in Close().
        consumer_sem_.Signal();
      } catch (...) {
        // There is nothing we called above that could potentially throw due to
        // user data.  So we treat reaching this point as a code-error condition.
        // Closing base_reader_ will trigger an exception in Next() in the main
        // thread when it checks that base_reader_->IsOpen().
        if (base_reader_->IsOpen()) {
          base_reader_->Close();
          delete base_reader_;
          base_reader_ = NULL;
        }
        consumer_sem_.Signal();
        return;
      }
    }
    static void run(SequentialTableReaderBackgroundImpl<Holder> *object) {
      object->RunInBackground();
    }
    virtual bool Done() const {
      return key_.empty();
    }
    virtual std::string Key() {
      if (key_.empty())
        KALDI_ERR << "Calling Key() at the wrong time.";
      return key_;
    }
    virtual T &Value() {
      if (key_.empty())
        KALDI_ERR << "Calling Value() at the wrong time.";
      return holder_.Value();
    }
    void SwapHolder(Holder *other_holder) {
      KALDI_ERR << "SwapHolder() should not be called on this class.";
    }
    virtual void FreeCurrent() {
      if (key_.empty())
        KALDI_ERR << "Calling FreeCurrent() at the wrong time.";
      // note: ideally a call to Value() should crash if you have just called
      // FreeCurrent().  For typical holders such as KaldiObjectHolder this will
      // happen inside the holder_.Value() call.  This won't be the case for all
      // holders, but it's not a great loss (just a missed opportunity to spot a
      // code error).
      holder_.Clear();
    }
    virtual void Next() {
      consumer_sem_.Wait();
      if (base_reader_ == NULL || !base_reader_->IsOpen())
        KALDI_ERR << "Error detected (likely code error) in background "
                  << "reader (',bg' option)";
      if (base_reader_->Done()) {
        // there is nothing else to read.
        key_ = "";
      } else {
        key_ = base_reader_->Key();
        base_reader_->SwapHolder(&holder_);
      }
      // this Signal() tells the producer thread, in the background,
      // that it's now safe to read the next value.
      producer_sem_.Signal();
    }
  
    // note: we can be sure that Close() won't be called twice, as the TableReader
    // object will delete this object after calling Close.
    virtual bool Close() {
      KALDI_ASSERT(base_reader_ != NULL && thread_.joinable());
      // wait until the producer thread is idle.
      consumer_sem_.Wait();
      bool ans = true;
      try {
        ans = base_reader_->Close();
      } catch (...) {
        ans = false;
      }
      delete base_reader_;
      // setting base_reader_ to NULL will cause the loop in the producer thread
      // to exit.
      base_reader_ = NULL;
      producer_sem_.Signal();
  
      thread_.join();
      return ans;
    }
    ~SequentialTableReaderBackgroundImpl() {
      if (base_reader_) {
        if (!Close()) {
          KALDI_ERR << "Error detected closing background reader "
                    << "(relates to ',bg' modifier)";
        }
      }
    }
   private:
    std::string key_;
    Holder holder_;
    // I couldn't figure out what to call these semaphores.  consumer_sem_ is the
    // one that the consumer (main thread) waits on; producer_sem_ is the one
    // that the producer (background thread) waits on.
    Semaphore consumer_sem_;
    Semaphore producer_sem_;
    std::thread thread_;
    SequentialTableReaderImplBase<Holder> *base_reader_;
  
  };
  
  template<class Holder>
  SequentialTableReader<Holder>::SequentialTableReader(const std::string
                                                       &rspecifier): impl_(NULL) {
    if (rspecifier != "" && !Open(rspecifier))
      KALDI_ERR << "Error constructing TableReader: rspecifier is " << rspecifier;
  }
  
  template<class Holder>
  bool SequentialTableReader<Holder>::Open(const std::string &rspecifier) {
    if (IsOpen())
      if (!Close())
        KALDI_ERR << "Could not close previously open object.";
    // now impl_ will be NULL.
  
    RspecifierOptions opts;
    RspecifierType wt = ClassifyRspecifier(rspecifier, NULL, &opts);
    switch (wt) {
      case kArchiveRspecifier:
        impl_ = new SequentialTableReaderArchiveImpl<Holder>();
        break;
      case kScriptRspecifier:
        impl_ = new SequentialTableReaderScriptImpl<Holder>();
        break;
      case kNoRspecifier: default:
        KALDI_WARN << "Invalid rspecifier " << rspecifier;
        return false;
    }
    if (!impl_->Open(rspecifier)) {
      delete impl_;
      impl_ = NULL;
      return false;  // sub-object will have printed warnings.
    }
    if (opts.background) {
      impl_ = new SequentialTableReaderBackgroundImpl<Holder>(
          impl_);
      if (!impl_->Open("")) {
        // the rxfilename is ignored in that Open() call.
        // It should only return false on code error.
        return false;
      }
    }
    return true;
  }
  
  template<class Holder>
  bool SequentialTableReader<Holder>::Close() {
    CheckImpl();
    bool ans = impl_->Close();
    delete impl_;  // We don't keep around empty impl_ objects.
    impl_ = NULL;
    return ans;
  }
  
  
  template<class Holder>
  bool SequentialTableReader<Holder>::IsOpen() const {
    return (impl_ != NULL);  // Because we delete the object whenever
    // that object is not open.  Thus, the IsOpen functions of the
    // Impl objects are not really needed.
  }
  
  template<class Holder>
  std::string SequentialTableReader<Holder>::Key() {
    CheckImpl();
    return impl_->Key();  // this call may throw if called wrongly in other ways,
    // e.g. eof.
  }
  
  
  template<class Holder>
  void SequentialTableReader<Holder>::FreeCurrent() {
    CheckImpl();
    impl_->FreeCurrent();
  }
  
  
  template<class Holder>
  typename SequentialTableReader<Holder>::T &
  SequentialTableReader<Holder>::Value() {
    CheckImpl();
    return impl_->Value();  // This may throw (if EnsureObjectLoaded() returned false you
                            // are safe.).
  }
  
  
  template<class Holder>
  void SequentialTableReader<Holder>::Next() {
    CheckImpl();
    impl_->Next();
  }
  
  template<class Holder>
  bool SequentialTableReader<Holder>::Done() {
    CheckImpl();
    return impl_->Done();
  }
  
  
  template<class Holder>
  SequentialTableReader<Holder>::~SequentialTableReader() {
    delete impl_;
    // Destructor of impl_ may throw.
  }
  
  
  
  template<class Holder> class TableWriterImplBase {
   public:
    typedef typename Holder::T T;
  
    virtual bool Open(const std::string &wspecifier) = 0;
  
    // Write returns true on success, false on failure, but
    // some errors may not be detected until we call Close().
    // It throws (via KALDI_ERR) if called wrongly.  We could
    // have just thrown on all errors, since this is what
    // TableWriter does; it was designed this way because originally
    // TableWriter::Write returned an exit status.
    virtual bool Write(const std::string &key, const T &value) = 0;
  
    // Flush will flush any archive; it does not return error status,
    //  any errors will be reported on the next Write or Close.
    virtual void Flush() = 0;
  
    virtual bool Close() = 0;
  
    virtual bool IsOpen() const = 0;
  
    // May throw on write error if Close was not called.
    virtual ~TableWriterImplBase() { }
  
    TableWriterImplBase() { }
   private:
    KALDI_DISALLOW_COPY_AND_ASSIGN(TableWriterImplBase);
  };
  
  
  // The implementation of TableWriter we use when writing directly
  // to an archive with no associated scp.
  template<class Holder>
  class TableWriterArchiveImpl: public TableWriterImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    virtual bool Open(const std::string &wspecifier) {
      switch (state_) {
        case kUninitialized:
          break;
        case kWriteError:
          KALDI_ERR << "Opening stream, already open with write error.";
        case kOpen: default:
          if (!Close())  // throw because this error may not have been previously
            // detected by the user.
            KALDI_ERR << "Opening stream, error closing previously open stream.";
      }
      wspecifier_ = wspecifier;
      WspecifierType ws = ClassifyWspecifier(wspecifier,
                                             &archive_wxfilename_,
                                             NULL,
                                             &opts_);
      KALDI_ASSERT(ws == kArchiveWspecifier);  // or wrongly called.
  
      if (output_.Open(archive_wxfilename_, opts_.binary, false)) {  // false
                                                        // means no binary header.
        state_ = kOpen;
        return true;
      } else {
        // stream will not be open.  User will report this error
        // (we return bool), so don't bother printing anything.
        state_ = kUninitialized;
        return false;
      }
    }
  
    virtual bool IsOpen() const {
      switch (state_) {
        case kUninitialized: return false;
        case kOpen: case kWriteError: return true;
        default: KALDI_ERR << "IsOpen() called on TableWriter in invalid state.";
      }
      return false;
    }
  
    // Write returns true on success, false on failure, but
    // some errors may not be detected till we call Close().
    virtual bool Write(const std::string &key, const T &value) {
      switch (state_) {
        case kOpen: break;
        case kWriteError:
          // user should have known from the last
          // call to Write that there was a problem.
          KALDI_WARN << "Attempting to write to invalid stream.";
          return false;
        case kUninitialized: default:
          KALDI_ERR << "Write called on invalid stream";
      }
      // state is now kOpen or kWriteError.
      if (!IsToken(key))  // e.g. empty string or has spaces...
        KALDI_ERR << "Using invalid key " << key;
      output_.Stream() << key << ' ';
      if (!Holder::Write(output_.Stream(), opts_.binary, value)) {
        KALDI_WARN << "Write failure to "
                   << PrintableWxfilename(archive_wxfilename_);
        state_ = kWriteError;
        return false;
      }
      if (state_ == kWriteError) return false;  // Even if this Write seems to
      // have succeeded, we fail because a previous Write failed and the archive
      // may be corrupted and unreadable.
  
      if (opts_.flush)
        Flush();
      return true;
    }
  
    // Flush will flush any archive; it does not return error status,
    //  any errors will be reported on the next Write or Close.
    virtual void Flush() {
      switch (state_) {
        case kWriteError: case kOpen:
          output_.Stream().flush();  // Don't check error status.
          return;
        default:
          KALDI_WARN << "Flush called on not-open writer.";
      }
    }
  
    virtual bool Close() {
      if (!this->IsOpen() || !output_.IsOpen())
        KALDI_ERR << "Close called on a stream that was not open."
                  << this->IsOpen() << ", " << output_.IsOpen();
      bool close_success = output_.Close();
      if (!close_success) {
        KALDI_WARN << "Error closing stream: wspecifier is " << wspecifier_;
        state_ = kUninitialized;
        return false;
      }
      if (state_ == kWriteError) {
        KALDI_WARN << "Closing writer in error state: wspecifier is "
                   << wspecifier_;
        state_ = kUninitialized;
        return false;
      }
      state_ = kUninitialized;
      return true;
    }
  
    TableWriterArchiveImpl(): state_(kUninitialized) {}
  
    // May throw on write error if Close was not called.
    virtual ~TableWriterArchiveImpl() {
      if (!IsOpen()) return;
      else if (!Close())
        KALDI_ERR << "At TableWriter destructor: Write failed or stream close "
                  << "failed: wspecifier is "<<  wspecifier_;
    }
  
   private:
    Output output_;
    WspecifierOptions opts_;
    std::string wspecifier_;
    std::string archive_wxfilename_;
    enum {               // is stream open?
      kUninitialized,    // no
      kOpen,             // yes
      kWriteError,       // yes
    } state_;
  };
  
  
  
  
  // The implementation of TableWriter we use when writing to
  // individual files (more generally, wxfilenames) specified
  // in an scp file that we read.
  
  // Note: the code for this class is similar to
  // RandomAccessTableReaderScriptImpl; try to keep them in sync.
  
  template<class Holder>
  class TableWriterScriptImpl: public TableWriterImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    TableWriterScriptImpl(): last_found_(0), state_(kUninitialized) {}
  
    virtual bool Open(const std::string &wspecifier) {
      switch (state_) {
        case kReadScript:
          KALDI_ERR << " Opening already open TableWriter: call Close first.";
        case kUninitialized: case kNotReadScript:
          break;
      }
      wspecifier_ = wspecifier;
      WspecifierType ws = ClassifyWspecifier(wspecifier,
                                             NULL,
                                             &script_rxfilename_,
                                             &opts_);
      KALDI_ASSERT(ws == kScriptWspecifier);  // or wrongly called.
      KALDI_ASSERT(script_.empty());  // no way it could be nonempty at this point.
  
      if (!ReadScriptFile(script_rxfilename_,
                           true,  // print any warnings
                           &script_)) {  // error reading script file or invalid
                                         // format
        state_ = kNotReadScript;
        return false;  // no need to print further warnings.  user gets the error.
      }
      std::sort(script_.begin(), script_.end());
      for (size_t i = 0; i+1 < script_.size(); i++) {
        if (script_[i].first.compare(script_[i+1].first) >= 0) {
          // script[i] not < script[i+1] in lexical order...
          KALDI_WARN << "Script file " << PrintableRxfilename(script_rxfilename_)
                     << " contains duplicate key " << script_[i].first;
          state_ = kNotReadScript;
          return false;
        }
      }
      state_ = kReadScript;
      return true;
    }
  
    virtual bool IsOpen() const {  return (state_ == kReadScript);  }
  
    virtual bool Close() {
      if (!IsOpen())
        KALDI_ERR << "Close() called on TableWriter that was not open.";
      state_ = kUninitialized;
      last_found_ = 0;
      script_.clear();
      return true;
    }
  
    // Write returns true on success, false on failure, but
    // some errors may not be detected till we call Close().
    virtual bool Write(const std::string &key, const T &value) {
      if (!IsOpen())
        KALDI_ERR << "Write called on invalid stream";
  
      if (!IsToken(key))  // e.g. empty string or has spaces...
        KALDI_ERR << "Using invalid key " << key;
  
      std::string wxfilename;
      if (!LookupFilename(key, &wxfilename)) {
        if (opts_.permissive) {
          return true;  // In permissive mode, it's as if we're writing to
                       // /dev/null for missing keys.
        } else {
          KALDI_WARN << "Script file "
                     << PrintableRxfilename(script_rxfilename_)
                     << " has no entry for key " <<key;
          return false;
        }
      }
      Output output;
      if (!output.Open(wxfilename, opts_.binary, false)) {
        // Open in the text/binary mode (on Windows) given by member var. "binary"
        // (obtained from wspecifier), but do not put the binary-mode header (it
        // will be written, if needed, by the Holder::Write function.)
        KALDI_WARN << "Failed to open stream: "
                   << PrintableWxfilename(wxfilename);
        return false;
      }
      if (!Holder::Write(output.Stream(), opts_.binary, value)
          || !output.Close()) {
        KALDI_WARN << "Failed to write data to "
                   << PrintableWxfilename(wxfilename);
  
        return false;
      }
      return true;
    }
  
    // Flush does nothing in this implementation, there is nothing to flush.
    virtual void Flush() { }
  
  
    virtual ~TableWriterScriptImpl() {
      // Nothing to do in destructor.
    }
  
   private:
    // Note: this function is almost the same as in
    // RandomAccessTableReaderScriptImpl.
    bool LookupFilename(const std::string &key, std::string *wxfilename) {
      // First, an optimization: if we're going consecutively, this will
      // make the lookup very fast.
      last_found_++;
      if (last_found_ < script_.size() && script_[last_found_].first == key) {
        *wxfilename = script_[last_found_].second;
        return true;
      }
      std::pair<std::string, std::string> pr(key, "");  // Important that ""
      // compares less than or equal to any string, so lower_bound points to the
      // element that has the same key.
      typedef typename std::vector<std::pair<std::string, std::string> >
                       ::const_iterator IterType;
      IterType iter = std::lower_bound(script_.begin(), script_.end(), pr);
      if (iter != script_.end() && iter->first == key) {
        last_found_ = iter - script_.begin();
        *wxfilename = iter->second;
        return true;
      } else {
        return false;
      }
    }
  
  
    WspecifierOptions opts_;
    std::string wspecifier_;
    std::string script_rxfilename_;
  
    // the script_ variable contains pairs of (key, filename), sorted using
    // std::sort.  This can be used with binary_search to look up filenames for
    // writing.  If this becomes inefficient we can use std::unordered_map (but I
    // suspect this wouldn't be significantly faster & would use more memory).
    // If memory becomes a problem here, the user should probably be passing
    // only the relevant part of the scp file rather than expecting us to get too
    // clever in the code.
    std::vector<std::pair<std::string, std::string> > script_;
    size_t last_found_;  // This is for an optimization used in LookupFilename.
  
    enum {
      kUninitialized,
      kReadScript,
      kNotReadScript,  // read of script failed.
    } state_;
  };
  
  
  // The implementation of TableWriter we use when writing directly
  // to an archive plus an associated scp.
  template<class Holder>
  class TableWriterBothImpl: public TableWriterImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    virtual bool Open(const std::string &wspecifier) {
      switch (state_) {
        case kUninitialized:
          break;
        case kWriteError:
          KALDI_ERR << "Opening stream, already open with write error.";
        case kOpen: default:
          if (!Close())  // throw because this error may not have been previously
                         // detected by user.
            KALDI_ERR << "Opening stream, error closing previously open stream.";
      }
      wspecifier_ = wspecifier;
      WspecifierType ws = ClassifyWspecifier(wspecifier,
                                             &archive_wxfilename_,
                                             &script_wxfilename_,
                                             &opts_);
      KALDI_ASSERT(ws == kBothWspecifier);  // or wrongly called.
      if (ClassifyWxfilename(archive_wxfilename_) != kFileOutput)
        KALDI_WARN << "When writing to both archive and script, the script file "
            "will generally not be interpreted correctly unless the archive is "
            "an actual file: wspecifier = " << wspecifier;
  
      if (!archive_output_.Open(archive_wxfilename_, opts_.binary, false)) {
        // false means no binary header.
        state_ = kUninitialized;
        return false;
      }
      if (!script_output_.Open(script_wxfilename_, false, false)) {  // first
        // false means text mode: script files always text-mode.   second false
        //  means don't write header (doesn't matter for text mode).
        archive_output_.Close();  // Don't care about status: error anyway.
        state_ = kUninitialized;
        return false;
      }
      state_ = kOpen;
      return true;
    }
  
    virtual bool IsOpen() const {
      switch (state_) {
        case kUninitialized: return false;
        case kOpen: case kWriteError: return true;
        default: KALDI_ERR << "IsOpen() called on TableWriter in invalid state.";
      }
      return false;
    }
  
    void MakeFilename(typename std::ostream::pos_type streampos,
                      std::string *output) const {
      std::ostringstream ss;
      ss << ':' << streampos;
      KALDI_ASSERT(ss.str() != ":-1");
      *output = archive_wxfilename_ + ss.str();
  
      // e.g. /some/file:12302.
      // Note that we warned if archive_wxfilename_ is not an actual filename;
      // the philosophy is we give the user rope and if they want to hang
      // themselves, with it, fine.
    }
  
    // Write returns true on success, false on failure, but
    // some errors may not be detected till we call Close().
    virtual bool Write(const std::string &key, const T &value) {
      switch (state_) {
        case kOpen: break;
        case kWriteError:
          // user should have known from the last
          // call to Write that there was a problem.  Warn about it.
          KALDI_WARN << "Writing to non-open TableWriter object.";
          return false;
        case kUninitialized: default:
          KALDI_ERR << "Write called on invalid stream";
      }
      // state is now kOpen or kWriteError.
      if (!IsToken(key))  // e.g. empty string or has spaces...
        KALDI_ERR << "Using invalid key " << key;
      std::ostream &archive_os = archive_output_.Stream();
      archive_os << key << ' ';
      typename std::ostream::pos_type archive_os_pos = archive_os.tellp();
      // position at start of Write() to archive.  We will record this in the
      // script file.
      std::string offset_rxfilename;  // rxfilename with offset into the archive,
      // e.g. some_archive_name.ark:431541423
      MakeFilename(archive_os_pos, &offset_rxfilename);
  
      // Write to the script file first.
      // The idea is that we want to get all the information possible into the
      // script file, to make it easier to unwind errors later.
      std::ostream &script_os = script_output_.Stream();
      script_output_.Stream() << key << ' ' << offset_rxfilename << '
  ';
  
      if (!Holder::Write(archive_output_.Stream(), opts_.binary, value)) {
        KALDI_WARN << "Write failure to"
                   << PrintableWxfilename(archive_wxfilename_);
        state_ = kWriteError;
        return false;
      }
  
      if (script_os.fail()) {
        KALDI_WARN << "Write failure to script file detected: "
                   << PrintableWxfilename(script_wxfilename_);
        state_ = kWriteError;
        return false;
      }
  
      if (archive_os.fail()) {
        KALDI_WARN << "Write failure to archive file detected: "
                   << PrintableWxfilename(archive_wxfilename_);
        state_ = kWriteError;
        return false;
      }
  
      if (state_ == kWriteError) return false;  // Even if this Write seems to
      // have succeeded, we fail because a previous Write failed and the archive
      // may be corrupted and unreadable.
  
      if (opts_.flush)
        Flush();
      return true;
    }
  
    // Flush will flush any archive; it does not return error status,
    //  any errors will be reported on the next Write or Close.
    virtual void Flush() {
      switch (state_) {
        case kWriteError: case kOpen:
          archive_output_.Stream().flush();  // Don't check error status.
          script_output_.Stream().flush();  // Don't check error status.
          return;
        default:
          KALDI_WARN << "Flush called on not-open writer.";
      }
    }
  
    virtual bool Close() {
      if (!this->IsOpen())
        KALDI_ERR << "Close called on a stream that was not open.";
      bool close_success = true;
      if (archive_output_.IsOpen())
        if (!archive_output_.Close()) close_success = false;
      if (script_output_.IsOpen())
        if (!script_output_.Close()) close_success = false;
      bool ans = close_success && (state_ != kWriteError);
      state_ = kUninitialized;
      return ans;
    }
  
    TableWriterBothImpl(): state_(kUninitialized) {}
  
    // May throw on write error if Close() was not called.
    // User can get the error status by calling Close().
    virtual ~TableWriterBothImpl() {
      if (!IsOpen()) return;
      else if (!Close())
        KALDI_ERR << "Write failed or stream close failed: "
                  << wspecifier_;
    }
  
   private:
    Output archive_output_;
    Output script_output_;
    WspecifierOptions opts_;
    std::string archive_wxfilename_;
    std::string script_wxfilename_;
    std::string wspecifier_;
    enum {               // is stream open?
      kUninitialized,    // no
      kOpen,             // yes
      kWriteError,       // yes
    } state_;
  };
  
  
  template<class Holder>
  TableWriter<Holder>::TableWriter(const std::string &wspecifier): impl_(NULL) {
    if (wspecifier != "" && !Open(wspecifier))
      KALDI_ERR << "Failed to open table for writing with wspecifier: " << wspecifier
                << ": errno (in case it's relevant) is: " << strerror(errno);
  }
  
  template<class Holder>
  bool TableWriter<Holder>::IsOpen() const {
    return (impl_ != NULL);
  }
  
  
  template<class Holder>
  bool TableWriter<Holder>::Open(const std::string &wspecifier) {
    if (IsOpen()) {
      if (!Close())  // call Close() yourself to suppress this exception.
        KALDI_ERR << "Failed to close previously open writer.";
    }
    KALDI_ASSERT(impl_ == NULL);
    WspecifierType wtype = ClassifyWspecifier(wspecifier, NULL, NULL, NULL);
    switch (wtype) {
      case kBothWspecifier:
        impl_ = new TableWriterBothImpl<Holder>();
        break;
      case kArchiveWspecifier:
        impl_ = new TableWriterArchiveImpl<Holder>();
        break;
      case kScriptWspecifier:
        impl_ = new TableWriterScriptImpl<Holder>();
        break;
      case kNoWspecifier: default:
        KALDI_WARN << "ClassifyWspecifier: invalid wspecifier " << wspecifier;
        return false;
    }
    if (impl_->Open(wspecifier)) {
      return true;
    } else {  // The class will have printed a more specific warning.
      delete impl_;
      impl_ = NULL;
      return false;
    }
  }
  
  template<class Holder>
  void TableWriter<Holder>::Write(const std::string &key,
                                  const T &value) const {
    CheckImpl();
    if (!impl_->Write(key, value))
      KALDI_ERR << "Error in TableWriter::Write";
    // More specific warning will have
    // been printed in the Write function.
  }
  
  template<class Holder>
  void TableWriter<Holder>::Flush() {
    CheckImpl();
    impl_->Flush();
  }
  
  template<class Holder>
  bool TableWriter<Holder>::Close() {
    CheckImpl();
    bool ans = impl_->Close();
    delete impl_;  // We don't keep around non-open impl_ objects
                   // [c.f. definition of IsOpen()]
    impl_ = NULL;
    return ans;
  }
  
  template<class Holder>
  TableWriter<Holder>::~TableWriter() {
    if (IsOpen() && !Close()) {
      KALDI_ERR << "Error closing TableWriter [in destructor].";
    }
  }
  
  
  // Types of RandomAccessTableReader:
  // In principle, we would like to have four types of RandomAccessTableReader:
  //  the 4 combinations  [scp, archive], [seekable, not-seekable],
  // where if something is seekable we only store a file offset.  However,
  // it seems sufficient for now to only implement two of these, in both
  // cases assuming it's not seekable so we never store file offsets and always
  // store either the scp line or the data in the archive.  The reasons are:
  // (1)
  // For scp files, storing the actual entry is not that much more expensive
  // than storing the file offsets (since the entries are just filenames), and
  // avoids a lot of fseek operations that might be expensive.
  // (2)
  // For archive files, there is no real reason, if you have the archive file
  // on disk somewhere, why you wouldn't access it via its associated scp.
  // [i.e. write it as ark, scp].  The main reason to read archives directly
  // is if they are part of a pipe, and in this case it's not seekable, so
  // we implement only this case.
  //
  // Note that we will rarely in practice have to keep in memory everything in
  // the archive, as long as things are only read once from the archive (the
  // "o, " or "once" option) and as long as we keep our keys in sorted order;
  // to take advantage of this we need the "s, " (sorted) option, so we would
  // read archives as e.g. "s, o, ark:-" (this is the rspecifier we would use if
  // it was the standard input and these conditions held).
  
  template<class Holder> class RandomAccessTableReaderImplBase {
   public:
    typedef typename Holder::T T;
  
    virtual bool Open(const std::string &rspecifier) = 0;
  
    virtual bool HasKey(const std::string &key) = 0;
  
    virtual const T &Value(const std::string &key) = 0;
  
    virtual bool Close() = 0;
  
    virtual ~RandomAccessTableReaderImplBase() {}
  };
  
  
  // Implementation of RandomAccessTableReader for a script file; for simplicity
  // we just read it in all in one go, as it's unlikely someone would generate
  // this from a pipe.  In principle we could read it on-demand as for the
  // archives, but this would probably be overkill.
  
  // Note: the code for this this class is similar to TableWriterScriptImpl:
  // try to keep them in sync.
  template<class Holder>
  class RandomAccessTableReaderScriptImpl:
        public RandomAccessTableReaderImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    RandomAccessTableReaderScriptImpl(): last_found_(0), state_(kUninitialized) {}
  
    virtual bool Open(const std::string &rspecifier) {
      switch (state_) {
        case kNotHaveObject: case kHaveObject: case kHaveRange:
          KALDI_ERR << " Opening already open RandomAccessTableReader:"
                       " call Close first.";
        case kUninitialized: case kNotReadScript:
          break;
      }
      rspecifier_ = rspecifier;
      RspecifierType rs = ClassifyRspecifier(rspecifier,
                                             &script_rxfilename_,
                                             &opts_);
      KALDI_ASSERT(rs == kScriptRspecifier);  // or wrongly called.
      KALDI_ASSERT(script_.empty());  // no way it could be nonempty at this point
  
      if (!ReadScriptFile(script_rxfilename_,
                          true,  // print any warnings
                          &script_)) {  // error reading script file or invalid
                                        // format
        state_ = kNotReadScript;
        return false;  // no need to print further warnings.  user gets the error.
      }
  
      rspecifier_ = rspecifier;
      // If opts_.sorted, the user has asserted that the keys are already sorted.
      // Although we could easily sort them, we want to let the user know of this
      // mistake.  This same mistake could have serious effects if used with an
      // archive rather than a script.
      if (!opts_.sorted)
        std::sort(script_.begin(), script_.end());
      for (size_t i = 0; i + 1 < script_.size(); i++) {
        if (script_[i].first.compare(script_[i+1].first) >= 0) {
          // script[i] not < script[i+1] in lexical order...
          bool same = (script_[i].first == script_[i+1].first);
          KALDI_WARN << "Script file " << PrintableRxfilename(script_rxfilename_)
                     << (same ? " contains duplicate key: " :
                         " is not sorted (remove s, option or add ns, option):"
                         " key is ") << script_[i].first;
          state_ = kNotReadScript;
          return false;
        }
      }
      state_ = kNotHaveObject;
      key_ = "";  // make sure we don't have a key set
      return true;
    }
  
    virtual bool IsOpen() const {
      return  (state_ == kNotHaveObject || state_ == kHaveObject ||
               state_ == kHaveRange);
    }
  
    virtual bool Close() {
      if (!IsOpen())
        KALDI_ERR << "Close() called on RandomAccessTableReader that was not"
                     " open.";
      holder_.Clear();
      range_holder_.Clear();
      state_ = kUninitialized;
      last_found_ = 0;
      script_.clear();
      key_ = "";
      range_ = "";
      data_rxfilename_ = "";
      // This cannot fail because any errors of a "global" nature would have been
      // detected when we did Open().  With archives it's different.
      return true;
    }
  
    virtual bool HasKey(const std::string &key) {
      bool preload = opts_.permissive;
      // In permissive mode, we have to check that we can read
      // the scp entry before we assert that the key is there.
      return HasKeyInternal(key, preload);
    }
  
  
    // Write returns true on success, false on failure, but
    // some errors may not be detected till we call Close().
    virtual const T&  Value(const std::string &key) {
      if (!HasKeyInternal(key, true)) // true == preload.
        KALDI_ERR << "Could not get item for key " << key
                  << ", rspecifier is " << rspecifier_ << " [to ignore this, "
                   << "add the p, (permissive) option to the rspecifier.";
      KALDI_ASSERT(key_ == key);
      if (state_ == kHaveObject) {
        return holder_.Value();
      } else {
        KALDI_ASSERT(state_ == kHaveRange);
        return range_holder_.Value();
      }
    }
  
    virtual ~RandomAccessTableReaderScriptImpl() { }
  
   private:
  
    // HasKeyInternal when called with preload == false just tells us whether the
    // key is in the scp.  With preload == true, which happens when the ,p
    // (permissive) option is given in the rspecifier (or when called from
    // Value()), it will also check that we can preload the object from disk
    // (loading from the rxfilename in the scp), and only return true if we can.
    // This function is called both from HasKey and from Value().
    virtual bool HasKeyInternal(const std::string &key, bool preload) {
      switch (state_) {
        case kUninitialized: case kNotReadScript:
          KALDI_ERR << "HasKey called on RandomAccessTableReader object that is"
                       " not open.";
        case kHaveObject:
          if (key == key_ && range_.empty())
            return true;
          break;
        case kHaveRange:
          if (key == key_)
            return true;
          break;
        case kNotHaveObject: default: break;
      }
      KALDI_ASSERT(IsToken(key));
      size_t key_pos = 0;
      if (!LookupKey(key, &key_pos)) {
        return false;
      } else {
        if (!preload) {
          return true;  // we have the key, and were not asked to verify that the
                        // object could be read.
        } else {  // preload specified, so we have to attempt to pre-load the
                  // object before returning.
          std::string data_rxfilename, range; // We will split
          // script_[key_pos].second (e.g. "1.ark:100[0:2]" into data_rxfilename
          // (e.g. "1.ark:100") and range (if any), e.g. "0:2".
          if (script_[key_pos].second[script_[key_pos].second.size()-1] == ']') {
            if(!ExtractRangeSpecifier(script_[key_pos].second,
                                      &data_rxfilename,
                                      &range)) {
              KALDI_ERR << "TableReader: failed to parse range in '"
                        << script_[key_pos].second << "'";
            }
          } else {
            data_rxfilename = script_[key_pos].second;
          }
          if (state_ == kHaveRange) {
            if (data_rxfilename_ == data_rxfilename && range_ == range) {
              // the odd situation where two keys had the same rxfilename and range:
              // just change the key and keep the object.
              key_ = key;
              return true;
            } else {
              range_holder_.Clear();
              state_ = kHaveObject;
            }
          }
          // OK, at this point the state will be kHaveObject or kNotHaveObject.
          if (state_ == kHaveObject) {
            if (data_rxfilename_ != data_rxfilename) {
              // clear out the object.
              state_ = kNotHaveObject;
              holder_.Clear();
            }
          }
          // At this point we can safely switch to the new key, data_rxfilename
          // and range, and we know that if we have an object, it will already be
          // the correct one.  The state is now kHaveObject or kNotHaveObject.
          key_ = key;
          data_rxfilename_ = data_rxfilename;
          range_ = range;
          if (state_ == kNotHaveObject) {
            // we need to read the object.
            if (!input_.Open(data_rxfilename)) {
              KALDI_WARN << "Error opening stream "
                         << PrintableRxfilename(data_rxfilename);
              return false;
            } else {
              if (holder_.Read(input_.Stream())) {
                state_ = kHaveObject;
              } else {
                KALDI_WARN << "Error reading object from "
                    "stream " << PrintableRxfilename(data_rxfilename);
                return false;
              }
            }
          }
          // At this point the state is kHaveObject.
          if (range.empty())
            return true;  // we're done: no range was requested.
          if (range_holder_.ExtractRange(holder_, range)) {
            state_ = kHaveRange;
            return true;
          } else {
            KALDI_WARN  << "Failed to load object from "
                        << PrintableRxfilename(data_rxfilename)
                        << "[" << range << "]";
            // leave state at kHaveObject.
            return false;
          }
        }
      }
    }
  
    // This function attempts to look up the key "key" in the sorted array
    // script_.  If it was found it returns true and puts the array offset into
    // 'script_offset'; otherwise it returns false.
    bool LookupKey(const std::string &key, size_t *script_offset) {
      // First, an optimization: if we're going consecutively, this will
      // make the lookup very fast.  Since we may call HasKey and then
      // Value(), which both may look up the key, we test if either the
      // current or next position are correct.
      if (last_found_ < script_.size() && script_[last_found_].first == key) {
        *script_offset = last_found_;
        return true;
      }
      last_found_++;
      if (last_found_ < script_.size() && script_[last_found_].first == key) {
        *script_offset = last_found_;
        return true;
      }
      std::pair<std::string, std::string> pr(key, "");  // Important that ""
      // compares less than or equal to any string, so lower_bound points to the
      // element that has the same key.
      typedef typename std::vector<std::pair<std::string, std::string> >
                       ::const_iterator IterType;
      IterType iter = std::lower_bound(script_.begin(), script_.end(), pr);
      if (iter != script_.end() && iter->first == key) {
        last_found_ = *script_offset = iter - script_.begin();
        return true;
      } else {
        return false;
      }
    }
  
  
    Input input_;  // Use the same input_ object for reading each file, in case
                   // the scp specifies offsets in an archive so we can keep the
                   // same file open.
    RspecifierOptions opts_;
    std::string rspecifier_;  // rspecifier used to open this object; used in
                              // debug messages
    std::string script_rxfilename_;  // rxfilename of script file that we read.
  
    std::string key_;  // The current key of the object that we have, but see the
                       // notes regarding states_ for more explanation of the
                       // semantics.
  
    Holder holder_;
    Holder range_holder_; // Holds the partial object corresponding to the object
                          // range specifier 'range_'. this is only used when
                          // 'range_' is specified.
    std::string range_; // range within which we read the object from holder_.
                        // If key_ is set, always correspond to the key.
    std::string data_rxfilename_;  // the rxfilename corresponding to key_,
                                   // always set when key_ is set.
  
  
    // the script_ variable contains pairs of (key, filename), sorted using
    // std::sort.  This can be used with binary_search to look up filenames for
    // writing.  If this becomes inefficient we can use std::unordered_map (but I
    // suspect this wouldn't be significantly faster & would use more memory).
    // If memory becomes a problem here, the user should probably be passing
    // only the relevant part of the scp file rather than expecting us to get too
    // clever in the code.
    std::vector<std::pair<std::string, std::string> > script_;
    size_t last_found_;  // This is for an optimization used in FindFilename.
  
    enum {
      //                   (*) is script_ set up?
      //                          (*) does holder_ contain an object?
      //                               (*) does range_holder_ contain and object?
      //
      //
      kUninitialized,  //    no    no    no
      kNotReadScript,  //    no    no    no
      kNotHaveObject,  //    yes   no    no
      kHaveObject,     //    yes   yes   no
      kHaveRange,      //    yes   yes   yes
  
      // If we are in a state where holder_ contains an object, it always contains
      // the object from 'key_', and the corresponding rxfilename is always
      // 'data_rxfilename_'.  If range_holder_ contains an object, it always
      // corresponds to the range 'range_' of the object in 'holder_', and always
      // corresponds to the current key.
    } state_;
  };
  
  
  
  
  // This is the base-class (with some implemented functions) for the
  // implementations of RandomAccessTableReader when it's an archive.  This
  // base-class handles opening the files, storing the state of the reading
  // process, and loading objects.  This is the only case in which we have
  // an intermediate class in the hierarchy between the virtual ImplBase
  // class and the actual Impl classes.
  // The child classes vary in the assumptions regarding sorting, etc.
  
  template<class Holder>
  class RandomAccessTableReaderArchiveImplBase:
        public RandomAccessTableReaderImplBase<Holder> {
   public:
    typedef typename Holder::T T;
  
    RandomAccessTableReaderArchiveImplBase(): holder_(NULL),
                                              state_(kUninitialized) { }
  
    virtual bool Open(const std::string &rspecifier) {
      if (state_ != kUninitialized) {
        if (!this->Close())  // call Close() yourself to suppress this exception.
          KALDI_ERR << "Error closing previous input.";
      }
      rspecifier_ = rspecifier;
      RspecifierType rs = ClassifyRspecifier(rspecifier, &archive_rxfilename_,
                                             &opts_);
      KALDI_ASSERT(rs == kArchiveRspecifier);
  
      // NULL means don't expect binary-mode header
      bool ans;
      if (Holder::IsReadInBinary())
        ans = input_.Open(archive_rxfilename_, NULL);
      else
        ans = input_.OpenTextMode(archive_rxfilename_);
      if (!ans) {  // header.
        KALDI_WARN << "Failed to open stream "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kUninitialized;  // Failure on Open
        return false;  // User should print the error message.
      } else {
        state_ = kNoObject;
      }
      return true;
    }
  
    // ReadNextObject() requires that the state be kNoObject,
    // and it will try read the next object.  If it succeeds,
    // it sets the state to kHaveObject, and
    // cur_key_ and holder_ have the key and value.  If it fails,
    // it sets the state to kError or kEof.
    void ReadNextObject() {
      if (state_ != kNoObject)
        KALDI_ERR << "ReadNextObject() called from wrong state.";
      // Code error somewhere in this class or a child class.
      std::istream &is = input_.Stream();
      is.clear();  // Clear any fail bits that may have been set... just in case
      // this happened in the Read function.
      is >> cur_key_;  // This eats up any leading whitespace and gets the string.
      if (is.eof()) {
        state_ = kEof;
        return;
      }
      if (is.fail()) {  // This shouldn't really happen, barring file-system
                        // errors.
        KALDI_WARN << "Error reading archive: rspecifier is " << rspecifier_;
        state_ = kError;
        return;
      }
      int c;
      if ((c = is.peek()) != ' ' && c != '\t' && c != '
  ') {  // We expect a
        // space ' ' after the key.
        // We also allow tab, just so we can read archives generated by scripts
        // that may not be fully aware of how this format works.
        KALDI_WARN << "Invalid archive file format: expected space after key "
                   <<cur_key_
                   <<", got character "
                   << CharToString(static_cast<char>(is.peek()))
                   << ", reading archive "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kError;
        return;
      }
      if (c != '
  ') is.get();  // Consume the space or tab.
      holder_ = new Holder;
      if (holder_->Read(is)) {
        state_ = kHaveObject;
        return;
      } else {
        KALDI_WARN << "Object read failed, reading archive "
                   << PrintableRxfilename(archive_rxfilename_);
        state_ = kError;
        delete holder_;
        holder_ = NULL;
        return;
      }
    }
  
    virtual bool IsOpen() const {
      switch (state_) {
        case kEof: case kError: case kHaveObject: case kNoObject: return true;
        case kUninitialized: return false;
        default: KALDI_ERR << "IsOpen() called on invalid object.";
          return false;
      }
    }
  
    // Called by the child-class virutal Close() functions; does the
    // shared parts of the cleanup.
    bool CloseInternal() {
      if (!this->IsOpen())
        KALDI_ERR << "Close() called on TableReader twice or otherwise wrongly.";
      if (input_.IsOpen())
        input_.Close();
      if (state_ == kHaveObject) {
        KALDI_ASSERT(holder_ != NULL);
        delete holder_;
        holder_ = NULL;
      } else {
        KALDI_ASSERT(holder_ == NULL);
      }
      bool ans = (state_ != kError);
      state_ = kUninitialized;
      if (!ans && opts_.permissive) {
        KALDI_WARN << "Error state detected closing reader.  "
                   << "Ignoring it because you specified permissive mode.";
        return true;
      }
      return ans;
    }
  
    ~RandomAccessTableReaderArchiveImplBase() {
      // The child class has the responsibility to call CloseInternal().
      KALDI_ASSERT(state_ == kUninitialized && holder_ == NULL);
    }
   private:
    Input input_;       // Input object for the archive
   protected:
    // The variables below are accessed by child classes.
  
    std::string cur_key_;   // current key (if state == kHaveObject).
    Holder *holder_;    // Holds the object we just read (if state == kHaveObject)
  
    std::string rspecifier_;
    std::string archive_rxfilename_;
    RspecifierOptions opts_;
  
    enum {  //  [The state of the reading process]        [does holder_ [is input_
      //                                                      have object]   open]
      kUninitialized,  // Uninitialized or closed                   no         no
      kNoObject,      // Do not have object in holder_              no         yes
      kHaveObject,    // Have object in holder_                     yes        yes
      kEof,           // End of file                                no         yes
      kError,         // Some kind of error-state in the reading.   no         yes
    } state_;
  };
  
  
  // RandomAccessTableReaderDSortedArchiveImpl (DSorted for "doubly sorted") is
  // the implementation for random-access reading of archives when both the
  // archive, and the calling code, are in sorted order (i.e. we ask for the keys
  // in sorted order).  This is when the s and cs options are both given.  It only
  // ever has to keep one object in memory.  It inherits from
  // RandomAccessTableReaderArchiveImplBase which implements the common parts of
  // RandomAccessTableReader that are used when it's an archive we're reading from
  
  template<class Holder>
  class RandomAccessTableReaderDSortedArchiveImpl:
        public RandomAccessTableReaderArchiveImplBase<Holder> {
    using RandomAccessTableReaderArchiveImplBase<Holder>::kUninitialized;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kHaveObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kNoObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kEof;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kError;
    using RandomAccessTableReaderArchiveImplBase<Holder>::state_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::opts_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::cur_key_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::holder_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::rspecifier_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::archive_rxfilename_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::ReadNextObject;
   public:
    typedef typename Holder::T T;
  
    RandomAccessTableReaderDSortedArchiveImpl() { }
  
    virtual bool Close() {
      // We don't have anything additional to clean up, so just
      // call generic base-class one.
      return this->CloseInternal();
    }
  
    virtual bool HasKey(const std::string &key) {
      return FindKeyInternal(key);
    }
    virtual const T & Value(const std::string &key) {
      if (!FindKeyInternal(key)) {
        KALDI_ERR << "Value() called but no such key " << key
                  << " in archive " << PrintableRxfilename(archive_rxfilename_);
      }
      KALDI_ASSERT(this->state_ == kHaveObject && key == this->cur_key_
                   && holder_ != NULL);
      return this->holder_->Value();
    }
  
    virtual ~RandomAccessTableReaderDSortedArchiveImpl() {
      if (this->IsOpen())
        if (!Close())  // more specific warning will already have been printed.
          // we are in some kind of error state & user did not find out by
          // calling Close().
          KALDI_ERR << "Error closing RandomAccessTableReader: rspecifier is "
                    << rspecifier_;
    }
   private:
    // FindKeyInternal tries to find the key by calling "ReadNextObject()"
    // as many times as necessary till we get to it.  It is called from
    // both FindKey and Value().
    bool FindKeyInternal(const std::string &key) {
      // First check that the user is calling us right: should be
      // in sorted order.  If not, error.
      if (!last_requested_key_.empty()) {
        if (key.compare(last_requested_key_) < 0) {  // key < last_requested_key_
          KALDI_ERR << "You provided the \"cs\" option "
                    << "but are not calling with keys in sorted order: "
                    << key << " < " << last_requested_key_ << ": rspecifier is "
                    << rspecifier_;
        }
      }
      // last_requested_key_ is just for debugging of order of calling.
      last_requested_key_ = key;
  
      if (state_ == kNoObject)
        ReadNextObject();  // This can only happen
        // once, the first time someone calls HasKey() or Value().  We don't
        // do it in the initializer to stop the program hanging too soon,
        // if reading from a pipe.
  
      if (state_ == kEof || state_ == kError) return false;
  
      if (state_ == kUninitialized)
        KALDI_ERR << "Trying to access a RandomAccessTableReader object that is"
                     " not open.";
  
      std::string last_key_;  // To check that
      // the archive we're reading is in sorted order.
      while (1) {
        KALDI_ASSERT(state_ == kHaveObject);
        int compare = key.compare(cur_key_);
        if (compare == 0) {  // key == key_
          return true;  // we got it..
        } else if (compare < 0) {  // key < cur_key_, so we already read past the
          // place where we want to be.  This implies that we will never find it
          // [due to the sorting etc., this means it just isn't in the archive].
          return false;
        } else {  // compare > 0, key > cur_key_.  We need to read further ahead.
          last_key_ = cur_key_;
          // read next object.. we have to set state to kNoObject first.
          KALDI_ASSERT(holder_ != NULL);
          delete holder_;
          holder_ = NULL;
          state_ = kNoObject;
          ReadNextObject();
          if (state_ != kHaveObject)
            return false;  // eof or read error.
          if (cur_key_.compare(last_key_) <= 0) {
            KALDI_ERR << "You provided the \"s\" option "
                      << " (sorted order), but keys are out of order or"
                         " duplicated: "
                      << last_key_ << " is followed by " << cur_key_
                      << ": rspecifier is " << rspecifier_;
          }
        }
      }
    }
  
    /// Last string provided to HasKey() or Value();
    std::string last_requested_key_;
  };
  
  // RandomAccessTableReaderSortedArchiveImpl is for random-access reading of
  // archives when the user specified the sorted (s) option but not the
  // called-sorted (cs) options.
  template<class Holder>
  class RandomAccessTableReaderSortedArchiveImpl:
        public RandomAccessTableReaderArchiveImplBase<Holder> {
    using RandomAccessTableReaderArchiveImplBase<Holder>::kUninitialized;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kHaveObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kNoObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kEof;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kError;
    using RandomAccessTableReaderArchiveImplBase<Holder>::state_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::opts_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::cur_key_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::holder_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::rspecifier_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::archive_rxfilename_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::ReadNextObject;
  
   public:
    typedef typename Holder::T T;
  
    RandomAccessTableReaderSortedArchiveImpl():
        last_found_index_(static_cast<size_t>(-1)),
        pending_delete_(static_cast<size_t>(-1)) { }
  
    virtual bool Close() {
      for (size_t i = 0; i < seen_pairs_.size(); i++)
        delete seen_pairs_[i].second;
      seen_pairs_.clear();
  
      pending_delete_ = static_cast<size_t>(-1);
      last_found_index_ = static_cast<size_t>(-1);
  
      return this->CloseInternal();
    }
    virtual bool HasKey(const std::string &key) {
      HandlePendingDelete();
      size_t index;
      bool ans = FindKeyInternal(key, &index);
      if (ans && opts_.once && seen_pairs_[index].second == NULL) {
        // Just do a check RE the once option. "&&opts_.once" is for
        // efficiency since this can only happen in that case.
        KALDI_ERR << "Error: HasKey called after Value() already called for "
                  << " that key, and once (o) option specified: rspecifier is "
                  << rspecifier_;
      }
      return ans;
    }
    virtual const T & Value(const std::string &key) {
      HandlePendingDelete();
      size_t index;
      if (!FindKeyInternal(key, &index)) {
        KALDI_ERR << "Value() called but no such key " << key
                  << " in archive " << PrintableRxfilename(archive_rxfilename_);
      }
      if (seen_pairs_[index].second == NULL) {  // can happen if opts.once_
        KALDI_ERR << "Error: Value() called more than once for key "
                  << key << " and once (o) option specified: rspecifier is "
                  << rspecifier_;
      }
      if (opts_.once)
        pending_delete_ = index;  // mark this index to be deleted on next call.
      return seen_pairs_[index].second->Value();
    }
    virtual ~RandomAccessTableReaderSortedArchiveImpl() {
      if (this->IsOpen())
        if (!Close())  // more specific warning will already have been printed.
          // we are in some kind of error state & user did not find out by
          // calling Close().
          KALDI_ERR << "Error closing RandomAccessTableReader: rspecifier is "
                    << rspecifier_;
    }
   private:
    void HandlePendingDelete() {
      const size_t npos = static_cast<size_t>(-1);
      if (pending_delete_ != npos) {
        KALDI_ASSERT(pending_delete_ < seen_pairs_.size());
        KALDI_ASSERT(seen_pairs_[pending_delete_].second != NULL);
        delete seen_pairs_[pending_delete_].second;
        seen_pairs_[pending_delete_].second = NULL;
        pending_delete_ = npos;
      }
    }
  
    // FindKeyInternal tries to find the key in the array "seen_pairs_".
    // If it is not already there, it reads ahead as far as necessary
    // to determine whether we have the key or not.  On success it returns
    // true and puts the index into the array seen_pairs_, into "index";
    // on failure it returns false.
    // It will leave the state as either kNoObject, kEof or kError.
    // FindKeyInternal does not do any checking about whether you are asking
    // about a key that has been already given (with the "once" option).
    // That is the user's responsibility.
  
    bool FindKeyInternal(const std::string &key, size_t *index) {
      // First, an optimization in case the previous call was for the
      // same key, and we found it.
      if (last_found_index_ < seen_pairs_.size()
         && seen_pairs_[last_found_index_].first == key) {
        *index = last_found_index_;
        return true;
      }
  
      if (state_ == kUninitialized)
        KALDI_ERR << "Trying to access a RandomAccessTableReader object that is"
                     " not open.";
  
      // Step one is to see whether we have to read ahead for the object..
      // Note, the possible states right now are kNoObject, kEof or kError.
      // We are never in the state kHaveObject except just after calling
      // ReadNextObject().
      bool looped = false;
      while (state_ == kNoObject &&
            (seen_pairs_.empty() || key.compare(seen_pairs_.back().first) > 0)) {
        looped = true;
        // Read this as:
        //  while ( the stream is potentially good for reading &&
        //        ([got no keys] || key > most_recent_key) ) { ...
        //     Try to read a new object.
        // Note that the keys in seen_pairs_ are ordered from least to greatest.
        ReadNextObject();
        if (state_ == kHaveObject) {  // Successfully read object.
          if (!seen_pairs_.empty() &&  // This is just a check.
             cur_key_.compare(seen_pairs_.back().first) <= 0) {
            // read the expression above as: !( cur_key_ > previous_key).
            // it means we are not in sorted order [the user specified that we
            // are, or we would not be using this implementation].
            KALDI_ERR << "You provided the sorted (s) option but keys in archive "
                      << PrintableRxfilename(archive_rxfilename_) << " are not "
                      << "in sorted order: " << seen_pairs_.back().first
                      << " is followed by " << cur_key_;
          }
          KALDI_ASSERT(holder_ != NULL);
          seen_pairs_.push_back(std::make_pair(cur_key_, holder_));
          holder_ = NULL;
          state_ = kNoObject;
        }
      }
      if (looped) {  // We only need to check the last element of the seen_pairs_
        // array, since we would not have read more after getting "key".
        if (!seen_pairs_.empty() && seen_pairs_.back().first == key) {
          last_found_index_ = *index = seen_pairs_.size() - 1;
          return true;
        } else {
          return false;
        }
      }
      // Now we have do an actual binary search in the seen_pairs_ array.
      std::pair<std::string, Holder*> pr(key, static_cast<Holder*>(NULL));
      typename std::vector<std::pair<std::string, Holder*> >::iterator
          iter = std::lower_bound(seen_pairs_.begin(), seen_pairs_.end(),
                                  pr, PairCompare());
      if (iter != seen_pairs_.end() &&
         key == iter->first) {
        last_found_index_ = *index = (iter - seen_pairs_.begin());
        return true;
      } else {
        return false;
      }
    }
  
    // These are the pairs of (key, object) we have read.  We keep all the keys we
    // have read but the actual objects (if they are stored with pointers inside
    // the Holder object) may be deallocated if once == true, and the Holder
    // pointer set to NULL.
    std::vector<std::pair<std::string, Holder*> > seen_pairs_;
    size_t last_found_index_;  // An optimization s.t. if FindKeyInternal called
    // twice with same key (as it often will), it doesn't have to do the key
    // search twice.
    size_t pending_delete_;  // If opts_.once == true, this is the index of
    // element of seen_pairs_ that is pending deletion.
    struct PairCompare {
      // PairCompare is the Less-than operator for the pairs of(key, Holder).
      // compares the keys.
      inline bool operator() (const std::pair<std::string, Holder*> &pr1,
                              const std::pair<std::string, Holder*> &pr2) {
        return  (pr1.first.compare(pr2.first) < 0);
      }
    };
  };
  
  
  
  // RandomAccessTableReaderUnsortedArchiveImpl is for random-access reading of
  // archives when the user does not specify the sorted (s) option (in this case
  // the called-sorted, or "cs" option, is ignored).  This is the least efficient
  // of the random access archive readers, in general, but it can be as efficient
  // as the others, in speed, memory and latency, if the "once" option is
  // specified and it happens that the keys of the archive are the same as the
  // keys the code is called with (to HasKey() and Value()), and in the same
  // order.  However, if you ask it for a key that's not present it will have to
  // read the archive till the end and store it all in memory.
  
  template<class Holder>
  class RandomAccessTableReaderUnsortedArchiveImpl:
        public RandomAccessTableReaderArchiveImplBase<Holder> {
    using RandomAccessTableReaderArchiveImplBase<Holder>::kUninitialized;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kHaveObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kNoObject;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kEof;
    using RandomAccessTableReaderArchiveImplBase<Holder>::kError;
    using RandomAccessTableReaderArchiveImplBase<Holder>::state_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::opts_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::cur_key_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::holder_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::rspecifier_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::archive_rxfilename_;
    using RandomAccessTableReaderArchiveImplBase<Holder>::ReadNextObject;
  
    typedef typename Holder::T T;
  
   public:
    RandomAccessTableReaderUnsortedArchiveImpl(): to_delete_iter_(map_.end()),
                                                  to_delete_iter_valid_(false) {
      map_.max_load_factor(0.5);  // make it quite empty -> quite efficient.
      // default seems to be 1.
    }
  
    virtual bool Close() {
      for (typename MapType::iterator iter = map_.begin();
          iter != map_.end();
          ++iter) {
        delete iter->second;
      }
      map_.clear();
      first_deleted_string_ = "";
      to_delete_iter_valid_ = false;
      return this->CloseInternal();
    }
  
    virtual bool HasKey(const std::string &key) {
      HandlePendingDelete();
      return FindKeyInternal(key, NULL);
    }
    virtual const T & Value(const std::string &key) {
      HandlePendingDelete();
      const T *ans_ptr = NULL;
      if (!FindKeyInternal(key, &ans_ptr))
        KALDI_ERR << "Value() called but no such key " << key
                  << " in archive " << PrintableRxfilename(archive_rxfilename_);
      return *ans_ptr;
    }
    virtual ~RandomAccessTableReaderUnsortedArchiveImpl() {
      if (this->IsOpen())
        if (!Close())  // more specific warning will already have been printed.
          // we are in some kind of error state & user did not find out by
          // calling Close().
          KALDI_ERR << "Error closing RandomAccessTableReader: rspecifier is "
                    << rspecifier_;
    }
   private:
    void HandlePendingDelete() {
      if (to_delete_iter_valid_) {
        to_delete_iter_valid_ = false;
        delete to_delete_iter_->second;  // Delete Holder object.
        if (first_deleted_string_.length() == 0)
          first_deleted_string_ = to_delete_iter_->first;
        map_.erase(to_delete_iter_);  // delete that element.
      }
    }
  
    // FindKeyInternal tries to find the key in the map "map_"
    // If it is not already there, it reads ahead either until it finds the
    // key, or until end of file.  If called with value_ptr == NULL,
    // it assumes it's called from HasKey() and just returns true or false
    // and doesn't otherwise have side effects.  If called with value_ptr !=
    // NULL, it assumes it's called from Value().  Thus, it will crash
    // if it cannot find the key.  If it can find it it puts its address in
    // *value_ptr, and if opts_once == true it will mark that element of the
    // map to be deleted.
  
    bool FindKeyInternal(const std::string &key, const T **value_ptr = NULL) {
      typename MapType::iterator iter = map_.find(key);
      if (iter != map_.end()) {  // Found in the map...
        if (value_ptr == NULL) {  // called from HasKey
          return true;  // this is all we have to do.
        } else {
          *value_ptr = &(iter->second->Value());
          if (opts_.once) {  // value won't be needed again, so mark
            // for deletion.
            to_delete_iter_ = iter;  // pending delete.
            KALDI_ASSERT(!to_delete_iter_valid_);
            to_delete_iter_valid_ = true;
          }
          return true;
        }
      }
      while (state_ == kNoObject) {
        ReadNextObject();
        if (state_ == kHaveObject) {  // Successfully read object.
          state_ = kNoObject;  // we are about to transfer ownership
          // of the object in holder_ to map_.
          // Insert it into map_.
          std::pair<typename MapType::iterator, bool> pr =
              map_.insert(typename MapType::value_type(cur_key_, holder_));
  
          if (!pr.second) {  // Was not inserted-- previous element w/ same key
            delete holder_;  // map was not changed, no ownership transferred.
            holder_ = NULL;
            KALDI_ERR << "Error in RandomAccessTableReader: duplicate key "
                      << cur_key_ << " in archive " << archive_rxfilename_;
          }
          holder_ = NULL;  // ownership transferred to map_.
          if (cur_key_ == key) {  // the one we wanted..
            if (value_ptr == NULL) {  // called from HasKey
              return true;
            } else {  // called from Value()
              *value_ptr = &(pr.first->second->Value());  // this gives us the
              // Value() from the Holder in the map.
              if (opts_.once) {  // mark for deletion, as won't be needed again.
                to_delete_iter_ = pr.first;
                KALDI_ASSERT(!to_delete_iter_valid_);
                to_delete_iter_valid_ = true;
              }
              return true;
            }
          }
        }
      }
      if (opts_.once && key == first_deleted_string_) {
        KALDI_ERR << "You specified the once (o) option but "
                  << "you are calling using key " << key
                  << " more than once: rspecifier is " << rspecifier_;
      }
      return false;  // We read the entire archive (or got to error state) and
      // didn't find it.
    }
  
    typedef unordered_map<std::string, Holder*, StringHasher>  MapType;
    MapType map_;
  
    typename MapType::iterator to_delete_iter_;
    bool to_delete_iter_valid_;
  
    std::string first_deleted_string_;  // keep the first string we deleted
    // from map_ (if opts_.once == true).  It's for an inexact spot-check that the
    // "once" option isn't being used incorrectly.
  };
  
  
  
  
  
  template<class Holder>
  RandomAccessTableReader<Holder>::RandomAccessTableReader(const
                                                         std::string &rspecifier):
      impl_(NULL) {
    if (rspecifier != "" && !Open(rspecifier))
      KALDI_ERR << "Error opening RandomAccessTableReader object "
          " (rspecifier is: " << rspecifier << ")";
  }
  
  template<class Holder>
  bool RandomAccessTableReader<Holder>::Open(const std::string &rspecifier) {
    if (IsOpen())
      KALDI_ERR << "Already open.";
    RspecifierOptions opts;
    RspecifierType rs = ClassifyRspecifier(rspecifier, NULL, &opts);
    switch (rs) {
      case kScriptRspecifier:
        impl_ = new RandomAccessTableReaderScriptImpl<Holder>();
        break;
      case kArchiveRspecifier:
        if (opts.sorted) {
          if (opts.called_sorted)  // "doubly" sorted case.
            impl_ = new RandomAccessTableReaderDSortedArchiveImpl<Holder>();
          else
            impl_ = new RandomAccessTableReaderSortedArchiveImpl<Holder>();
        } else {
          impl_ = new RandomAccessTableReaderUnsortedArchiveImpl<Holder>();
        }
        break;
      case kNoRspecifier: default:
        KALDI_WARN << "Invalid rspecifier: "
                   << rspecifier;
        return false;
    }
    if (!impl_->Open(rspecifier)) {
      // A warning will already have been printed.
      delete impl_;
      impl_ = NULL;
      return false;
    }
    return true;
  }
  
  template<class Holder>
  bool RandomAccessTableReader<Holder>::HasKey(const std::string &key) {
    CheckImpl();
    if (!IsToken(key))
      KALDI_ERR << "Invalid key \"" << key << '"';
    return impl_->HasKey(key);
  }
  
  
  template<class Holder>
  const typename RandomAccessTableReader<Holder>::T&
  RandomAccessTableReader<Holder>::Value(const std::string &key) {
    CheckImpl();
    return impl_->Value(key);
  }
  
  template<class Holder>
  bool RandomAccessTableReader<Holder>::Close() {
    CheckImpl();
    bool ans =impl_->Close();
    delete impl_;
    impl_ = NULL;
    return ans;
  }
  
  template<class Holder>
  RandomAccessTableReader<Holder>::~RandomAccessTableReader() {
    if (IsOpen() && !Close())  // call Close() yourself to stop this being thrown.
      KALDI_ERR << "failure detected in destructor.";
  }
  
  template<class Holder>
  void SequentialTableReader<Holder>::CheckImpl() const {
    if (!impl_) {
      KALDI_ERR << "Trying to use empty SequentialTableReader (perhaps you "
                << "passed the empty string as an argument to a program?)";
    }
  }
  
  template<class Holder>
  void RandomAccessTableReader<Holder>::CheckImpl() const {
    if (!impl_) {
      KALDI_ERR << "Trying to use empty RandomAccessTableReader (perhaps you "
                << "passed the empty string as an argument to a program?)";
    }
  }
  
  template<class Holder>
  void TableWriter<Holder>::CheckImpl() const {
    if (!impl_) {
      KALDI_ERR << "Trying to use empty TableWriter (perhaps you "
                << "passed the empty string as an argument to a program?)";
    }
  }
  
  template<class Holder>
  RandomAccessTableReaderMapped<Holder>::RandomAccessTableReaderMapped(
      const std::string &table_rxfilename,
      const std::string &utt2spk_rxfilename):
      reader_(table_rxfilename), token_reader_(table_rxfilename.empty() ? "" :
                                               utt2spk_rxfilename),
      utt2spk_rxfilename_(utt2spk_rxfilename) { }
  
  template<class Holder>
  bool RandomAccessTableReaderMapped<Holder>::Open(
      const std::string &table_rxfilename,
      const std::string &utt2spk_rxfilename) {
    if (reader_.IsOpen()) reader_.Close();
    if (token_reader_.IsOpen()) token_reader_.Close();
    KALDI_ASSERT(!table_rxfilename.empty());
    if (!reader_.Open(table_rxfilename)) return false;  // will have printed
    // warning internally, probably.
    if (!utt2spk_rxfilename.empty()) {
      if (!token_reader_.Open(utt2spk_rxfilename)) {
        reader_.Close();
        return false;
      }
    }
    return true;
  }
  
  
  template<class Holder>
  bool RandomAccessTableReaderMapped<Holder>::HasKey(const std::string &utt) {
    // We don't check IsOpen, we let the call go through to the member variable
    // (reader_), which will crash with a more informative error message than
    // we can give here, as we don't any longer know the rxfilename.
    if (token_reader_.IsOpen()) {  // We need to map the key from utt to spk.
      if (!token_reader_.HasKey(utt))
        KALDI_ERR << "Attempting to read key " << utt << ", which is not present "
                  << "in utt2spk map or similar map being read from "
                  << PrintableRxfilename(utt2spk_rxfilename_);
      const std::string &spk = token_reader_.Value(utt);
      return reader_.HasKey(spk);
    } else {
      return reader_.HasKey(utt);
    }
  }
  
  template<class Holder>
  const typename Holder::T& RandomAccessTableReaderMapped<Holder>::Value(
      const std::string &utt) {
    if (token_reader_.IsOpen()) {  // We need to map the key from utt to spk.
      if (!token_reader_.HasKey(utt))
        KALDI_ERR << "Attempting to read key " << utt << ", which is not present "
                  << "in utt2spk map or similar map being read from "
                  << PrintableRxfilename(utt2spk_rxfilename_);
      const std::string &spk = token_reader_.Value(utt);
      return reader_.Value(spk);
    } else {
      return reader_.Value(utt);
    }
  }
  
  
  
  /// @}
  
  }  // end namespace kaldi
  
  
  
  #endif  // KALDI_UTIL_KALDI_TABLE_INL_H_