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