dxfile.h

Go to the documentation of this file.

// Copyright (C) 2013-2016 DNAnexus, Inc.
//
// This file is part of dx-toolkit (DNAnexus platform client libraries).
//
//   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
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
//   License for the specific language governing permissions and limitations
//   under the License.

#ifndef DXCPP_BINDINGS_DXFILE_H
#define DXCPP_BINDINGS_DXFILE_H

#include <fstream>
#include <sstream>
#include <boost/thread.hpp>
#include "../bqueue.h"
#include "../bindings.h"
#include "../utils.h"

namespace dx {

  class DXFile: public DXDataObject {
   private:
    dx::JSON describe_(const std::string &s)const{return fileDescribe(dxid_,s);}
    void addTypes_(const std::string &s)const{fileAddTypes(dxid_,s);}
    void removeTypes_(const std::string &s)const{fileRemoveTypes(dxid_,s);}
    dx::JSON getDetails_(const std::string &s)const{return fileGetDetails(dxid_,s);}
    void setDetails_(const std::string &s)const{fileSetDetails(dxid_,s);}
    void setVisibility_(const std::string &s)const{fileSetVisibility(dxid_,s);}
    void rename_(const std::string &s)const{fileRename(dxid_,s);}
    void setProperties_(const std::string &s)const{fileSetProperties(dxid_,s);}
    void addTags_(const std::string &s)const{fileAddTags(dxid_,s);}
    void removeTags_(const std::string &s)const{fileRemoveTags(dxid_,s);}
    void close_(const std::string &s)const{fileClose(dxid_,s);}
    dx::JSON listProjects_(const std::string &s)const{return fileListProjects(dxid_,s);}

    // For async write() ///////////////////
    void joinAllWriteThreads_();
    void writeChunk_();
    void createWriteThreads_();
   
    // For linear query ///////////////////////////////////////////////
    void readChunk_() const;
    void getChunkHttp_(int64_t start, int64_t end, std::string& result) const;

    bool hasAnyPartBeenUploaded;

    int64_t pos_;

    int64_t gcount_;

    int64_t file_length_;

    std::stringstream buffer_;

    int cur_part_;

    bool eof_;

    mutable bool is_closed_;
    
    void reset_data_processing_();
    void reset_config_variables_();
    void reset_everything_();
    void copy_config_variables_(const DXFile &to_copy);
    void init_internals_();

    int64_t max_buf_size_;
    int max_write_threads_;

    // To allow interleaving (without compiler optimization possibly changing order)
    // we use std::atomic (a c++11 feature)
    // Ref https://parasol.tamu.edu/bjarnefest/program/boehm-slides.pdf (page 7)
    // Update: Since CLang does not support atomics yet, we are using locking 
    //         mechanism with alongwith volatile
    volatile int countThreadsWaitingOnConsume, countThreadsNotWaitingOnConsume;
    boost::mutex countThreadsMutex;
    std::vector<boost::thread> writeThreads;
    static const int DEFAULT_WRITE_THREADS = 5;
    static const int64_t DEFAULT_BUFFER_MAXSIZE = 100 * 1024 * 1024; // 100 MB
    BlockingQueue<std::pair<std::string, int> > uploadPartRequestsQueue;
    
    // For linear query
    mutable std::map<int64_t, std::string> lq_results_;
    mutable int64_t lq_chunk_limit_;
    mutable int64_t lq_query_start_;
    mutable int64_t lq_query_end_;
    mutable unsigned lq_max_chunks_;
    mutable int64_t lq_next_result_;
    mutable std::string lq_url;
    mutable dx::JSON lq_headers;
    mutable std::vector<boost::thread> lq_readThreads_;
    mutable boost::mutex lq_results_mutex_, lq_query_start_mutex_;

   public:

    DXFile(): DXDataObject() {
      reset_everything_(); 
    }

    DXFile(const DXFile& to_copy) : DXDataObject() {
      reset_everything_();
      setIDs(to_copy.dxid_, to_copy.proj_);
      copy_config_variables_(to_copy);
    }
    
    DXFile(const char *dxid, const char *proj=NULL): DXDataObject() {
      reset_everything_();
      setIDs(std::string(dxid), (proj == NULL) ? config::CURRENT_PROJECT() : std::string(proj));
    }
   
    DXFile(const std::string &dxid, const std::string &proj=config::CURRENT_PROJECT()): DXDataObject() {
      reset_everything_();
      setIDs(dxid, proj);
    }
    
    DXFile(const dx::JSON &dxlink): DXDataObject() {
      reset_everything_();
      setIDs(dxlink);
    }

    DXFile& operator=(const DXFile& to_copy) {
      if (this == &to_copy)
        return *this;

      this->setIDs(to_copy.dxid_, to_copy.proj_); // setIDs() will call reset_data_processing_() & init_internals_()
      this->copy_config_variables_(to_copy);
      return *this;
    }

    ~DXFile() {
      flush();
      stopLinearQuery();
    }
    // File-specific functions

    void setIDs(const std::string &dxid, const std::string &proj=config::CURRENT_PROJECT());
    
    void setIDs(const char *dxid, const char *proj = NULL);

    void setIDs(const dx::JSON &dxlink);

    void create(const std::string &media_type="",
                const dx::JSON &data_obj_fields=dx::JSON(dx::JSON_OBJECT));
    
    int64_t getMaxBufferSize() const {
      return max_buf_size_;
    }

    void setMaxBufferSize(const int64_t buf_size) {
      if (buf_size < (5 * 1024 * 1024)) { 
        throw DXFileError("Maximum buffer size for DXFile must be >= 5242880 (5MB)");
      }
      max_buf_size_ = buf_size;
    }
    
    int getNumWriteThreads() const {
      return max_write_threads_;
    }

    void setNumWriteThreads(const int numThreads) {
      max_write_threads_ = numThreads;
    }

    void read(char* ptr, int64_t n);

    int64_t gcount() const;

    bool eof() const;

    void seek(const int64_t pos);

    void flush();

    void write(const char* ptr, int64_t n);

    void write(const std::string &data);

    void uploadPart(const std::string &data, const int index=-1);

    void uploadPart(const char* ptr, int64_t n, const int index=-1);

    bool is_open() const;

    bool is_closed() const;

    void close(const bool block=false);

    void waitOnClose() const;

    void startLinearQuery(const int64_t start_byte=0,
                          const int64_t num_bytes=-1,
                          const int64_t chunk_size=10*1024*1024,
                          const unsigned max_chunks=20,
                          const unsigned thread_count=5) const;

    void stopLinearQuery() const;

    bool getNextChunk(std::string &chunk) const;

    static DXFile openDXFile(const std::string &dxid);

    static DXFile newDXFile(const std::string &media_type="",
                            const dx::JSON &data_obj_fields=
                            dx::JSON(dx::JSON_OBJECT));

    static void downloadDXFile(const std::string &dxid,
                               const std::string &filename,
                               int64_t chunksize=1048576);

    static DXFile uploadLocalFile(const std::string &filename,
                                  const std::string &media_type="",
                                  const dx::JSON &data_obj_fields=
                                  dx::JSON(dx::JSON_OBJECT),
                                  bool waitForClose=false);

    DXFile clone(const std::string &dest_proj_id,
                 const std::string &dest_folder="/") const;

  };
}
#endif