Merge pull request #172 from tobiasschuerg/feat/stream_batch

feat: streaming write

Author: vlastahajek

CHANGELOG.md

@@ -3,6 +3,8 @@
 ### Features
  - [167](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/167) - Added `InfluxDBClient::writeRecord(const char *record)`.
  - [167](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/167) - Added possibility to disable retrying by setting `maxRetryAttempts` to zero: `client.setWriteOptions(WriteOptions().maxRetryAttempts(0));` 
+ - [172](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/170) - Added directly streaming batch for write. It can be enable by `InfluxDBClient::setStreamWrite(bool enable = true)`. Writing by streaming lines of batch saves RAM as it sends data without allocating a buffer. On the other hand, this way of writing is about half times slower than the classic way, when allocating the buffer for writing the whole batch. 
+  - [172](https://github.com/tobiasschuerg/InfluxDB-Client-for-Arduino/pull/170) - Allowing larger batch size, > 255.
 
 ## 3.9.0 [2021-09-17]
 ### Features

README.md

@@ -23,6 +23,8 @@ This library doesn't support using those devices as a peripheral.
     - [Timestamp](#timestamp)
     - [Configure Time](#configure-time)
     - [Batch Size](#batch-size)
+    - [Large Batch Size](#large-batch-size)
+    - [Write Modes](#write-modes)
   - [Buffer Handling and Retrying](#buffer-handling-and-retrying)
   - [Write Options](#write-options)
   - [HTTP Options](#http-options)
@@ -176,7 +178,7 @@ client.setWriteOptions(WriteOptions().writePrecision(WritePrecision::MS));
 When a write precision is configured, the client will automatically assign the current time to the timestamp of each written point which doesn't have a timestamp assigned.
 
 If you want to manage timestamp on your own, there are several ways to set the timestamp explicitly.
-- `setTime(WritePrecision writePrecision)` - Sets the timestamp to the actual time in the desired precision
+- `setTime(WritePrecision writePrecision)` - Sets the timestamp to the actual time in the desired precision. The same precision must set in WriteOptions.
 - `setTime(unsigned long long timestamp)` -  Sets the timestamp to an offset since the epoch. Correct precision must be set InfluxDBClient::setWriteOptions.
 - `setTime(String timestamp)` - Sets the timestamp to an offset since the epoch. Correct precision must be set InfluxDBClient::setWriteOptions.
 
@@ -229,13 +231,13 @@ void timeSync(const char *tzInfo, const char* ntpServer1, const char* ntpServer2
 Setting batch size depends on data gathering and DB updating strategy.
 
 If data is written in short periods (seconds), the batch size should be set according to your expected write periods and update frequency requirements.
-For example, if you would like to see updates (on the dashboard or in processing) each minute and you are measuring a single value (1 point) every 10s (6 points per minute), the batch size should be 6. If it is sufficient to update each hour and you are creating 1 point each minute, your batch size should be 60. The maximum recommended batch size is 200. Maximum batch size depends on the RAM of the device (80KB for ESP8266 and 512KB for ESP32).
+For example, if you would like to see updates (on the dashboard or in processing) each minute and you are measuring a single value (1 point) every 10s (6 points per minute), the batch size should be 6. If it is sufficient to update each hour and you are creating 1 point each minute, your batch size should be 60. 
 
 In cases where the data should be written in longer periods and gathered data consists of several points, the batch size should be set to the expected number of points to be gathered.
 
 To set the batch size we use `WriteOptions` object and  [setWriteOptions](#write-options) function:
 ```cpp
-// Enable messages batching
+// Enable lines batching
 client.setWriteOptions(WriteOptions().batchSize(10));
 ```
 Writing the point will add a point to the underlying buffer until the batch size is reached:
@@ -257,6 +259,28 @@ if(!client.writePoint(point10)) {
 
 In case cases where the number of points is not always the same, set the batch size to the maximum number of points and use the `flushBuffer()` function to force writing to the database. See [Buffer Handling](#buffer-handling-and-retrying) for more details.
 
+### Large batch size
+The maximum batch size depends on the available RAM of the device (~45KB for ESP8266 and ~260KB for ESP32). Larger batch size, >100 for ESP8255, >2000 for ESP32, must be chosen carefully to not crash the app with out of memory error. The Stream write mode must be used, see [Write Modes](#write-modes)
+
+Always determine your typical line length using `client.pointToLineProtocol(point).length()`. For example, ESP32 can handle 2048 lines with an average length of 69. When the length of line or batch size is increased, the device becomes unstable, even there is more than 76k, it cannot send data or even crashes. ESP8266 handles successfully 330 of such lines. 
+
+:warning: Thoroughly test your app when using large batch files. 
+
+### Write Modes
+Client has two modes of writing:
+ - Buffer (default)
+ - Stream
+
+Writing is performed the way that client keeps written lines (points) separately and when a batch is completed, it allocates a data buffer for sending to a server via WiFi Client.
+This is the fastest way to write data but requires some amount of free memory. Thus a big batch size cannot be used.
+
+Another way of writing is *stream write*. 
+```cpp
+  // Enables stream write
+  client.setStreamWrite(true);
+```
+In this mode client continuously streams lines from batch to WiFi Client. No buffer allocation. As lines are allocated separately, it avoids problems with max allocable block size. The downside is, that writing is about 50% slower than in the Buffer mode.
+
 ## Buffer Handling and Retrying
 InfluxDB contains an underlying buffer for handling writing in batches and automatic retrying on server back-pressure and connection failure.
 
@@ -300,12 +324,15 @@ Writing points can be controlled via `WriteOptions`, which is set in the `setWri
 | batchSize | `1` | Number of points that will be written to the database at once |
 | bufferSize | `5` | Maximum number of points in buffer. Buffer contains new data that will be written to the database and also data that failed to be written due to network failure or server overloading |
 | flushInterval | `60` | Maximum time(in seconds) data will be held in buffer before points are written to the db |
+| retryInterval | `5` | Default retry interval in sec, if not sent by server. Value `0` disables retrying |
+| maxRetryInterval | `300` |  Maximum retry interval in sec |
+| maxRetryAttempts | `3` | Maximum count of retry attempts of failed writes |
 
 ## HTTP Options
 `HTTPOptions` controls some aspects of HTTP communication and they are set via `setHTTPOptions` function:
 | Parameter | Default Value | Meaning |
 |-----------|---------------|---------|
-| reuseConnection | `false` | Whether HTTP connection should be kept open after initial communication. Usable for frequent writes/queries. |
+| connectionReuse | `false` | Whether HTTP connection should be kept open after initial communication. Usable for frequent writes/queries. |
 | httpReadTimeout | `5000` | Timeout (ms) for reading server response |
 
 ## Secure Connection

src/BucketsClient.h

@@ -97,10 +97,8 @@ friend class E2ETest;
     // Returns true if a bucket exists
     bool checkBucketExists(const char *bucketName);
     // Returns a Bucket instance if a bucket is found.
-    // Returned instance must be manually deleted at the end of usage.
     Bucket findBucket(const char *bucketName);
     // Creates a bucket with given name and optional retention policy. 0 means infinite.
-    // Returned instance must be manually deleted at the end of usage.
     Bucket createBucket(const char *bucketName, uint32_t expiresSec = 0);
     // Delete a bucket with given id. Use findBucket to get a bucket with id.
     bool deleteBucket(const char *id);

src/HTTPService.cpp

@@ -159,6 +159,18 @@ bool HTTPService::doPOST(const char *url, const char *data, const char *contentT
   return afterRequest(expectedCode, cb);
 }
 
+bool HTTPService::doPOST(const char *url, Stream *stream, const char *contentType, int expectedCode, httpResponseCallback cb) {
+  INFLUXDB_CLIENT_DEBUG("[D] POST request - %s, data: %dbytes, type %s\n", url, stream->available(), contentType);
+  if(!beforeRequest(url)) {
+    return false;
+  }
+  if(contentType) {
+    _httpClient->addHeader(F("Content-Type"), FPSTR(contentType));
+  }
+  _lastStatusCode = _httpClient->sendRequest("POST", stream, stream->available());
+  return afterRequest(expectedCode, cb);
+}
+
 bool HTTPService::doGET(const char *url, int expectedCode, httpResponseCallback cb) {
   INFLUXDB_CLIENT_DEBUG("[D] GET request - %s\n", url);
   if(!beforeRequest(url)) {

src/HTTPService.h

@@ -113,6 +113,8 @@ friend class Test;
     HTTPOptions &getHTTPOptions() { return _httpOptions; }
     // Performs HTTP POST by sending data. On success calls response call back  
     bool doPOST(const char *url, const char *data, const char *contentType, int expectedCode, httpResponseCallback cb);
+    // Performs HTTP POST by sending stream. On success calls response call back  
+    bool doPOST(const char *url, Stream *stream, const char *contentType, int expectedCode, httpResponseCallback cb);
     // Performs HTTP GET. On success calls response call back    
     bool doGET(const char *url, int expectedCode, httpResponseCallback cb);
     // Performs HTTP DELETE. On success calls response call back    

src/InfluxDbClient.cpp

@@ -50,6 +50,7 @@ static String precisionToString(WritePrecision precision, uint8_t version = 2) {
     }
 }
 
+
 InfluxDBClient::InfluxDBClient() { 
    resetBuffer();
 }
@@ -122,6 +123,9 @@ bool InfluxDBClient::init() {
 
 InfluxDBClient::~InfluxDBClient() {
      if(_writeBuffer) {
+        for(int i=0;i<_writeBufferSize;i++) {
+            delete _writeBuffer[i];
+        }
         delete [] _writeBuffer;
         _writeBuffer = nullptr;
         _bufferPointer = 0;
@@ -294,15 +298,34 @@ bool InfluxDBClient::writePoint(Point & point) {
     return false;
 }
 
+InfluxDBClient::Batch::Batch(uint16_t size):_size(size) {  
+    buffer = new char*[size]; 
+    for(int i=0;i< _size; i++) {
+        buffer[i] = nullptr; 
+    }
+}
+
+
+InfluxDBClient::Batch::~Batch() { 
+    clear();
+    delete [] buffer; 
+    buffer = nullptr;
+}
+
+void InfluxDBClient::Batch::clear() {
+    for(int i=0;i< _size; i++) {
+        free(buffer[i]);
+        buffer[i] = nullptr; 
+    }
+}
+
 bool InfluxDBClient::Batch::append(const char *line) {
     if(pointer == _size) {
         //overwriting, clean buffer
-        for(int i=0;i< _size; i++) {
-            buffer[i] = (const char *)nullptr; 
-        }
+        clear();
         pointer = 0;
     } 
-    buffer[pointer] = line;
+    buffer[pointer] = strdup(line);
     ++pointer;
     return isFull();
 }
@@ -311,7 +334,7 @@ char * InfluxDBClient::Batch::createData() {
      int length = 0; 
      char *buff = nullptr;
      for(int c=0; c < pointer; c++) {
-        length += buffer[c].length();
+        length += strlen(buffer[c]);
         yield();
     }
     //create buffer for all lines including new line char and terminating char
@@ -320,7 +343,7 @@ char * InfluxDBClient::Batch::createData() {
         if(buff) {
             buff[0] = 0;
             for(int c=0; c < pointer; c++) {
-                strcat(buff+strlen(buff), buffer[c].c_str());
+                strcat(buff+strlen(buff), buffer[c]);
                 strcat(buff+strlen(buff), "\n");
                 yield();
             }
@@ -402,7 +425,6 @@ bool InfluxDBClient::flushBufferInternal(bool flashOnlyFull) {
     bool success = true;
     // send all batches, It could happen there was long network outage and buffer is full
     while(_writeBuffer[_batchPointer] && (!flashOnlyFull ||  _writeBuffer[_batchPointer]->isFull())) {
-        data = _writeBuffer[_batchPointer]->createData();
         if(!_writeBuffer[_batchPointer]->isFull() && _writeBuffer[_batchPointer]->retryCount == 0 ) { //do not increase pointer in case of retrying
             // points will be written so increase _bufferPointer as it happen when buffer is flushed when is full
             if(++_bufferPointer == _writeBufferSize) {
@@ -411,9 +433,15 @@ bool InfluxDBClient::flushBufferInternal(bool flashOnlyFull) {
         }
 
         INFLUXDB_CLIENT_DEBUG("[D] Writing batch, batchpointer: %d, size %d\n", _batchPointer, _writeBuffer[_batchPointer]->pointer);
-        if(data) {
-            int statusCode = postData(data);
-            delete [] data;
+        if(!_writeBuffer[_batchPointer]->isEmpty()) {
+            int statusCode = 0;
+            if(_streamWrite) {
+                statusCode = postData(_writeBuffer[_batchPointer]);
+            } else {
+                data = _writeBuffer[_batchPointer]->createData();
+                statusCode = postData(data);
+                delete [] data;
+            }
             // retry on unsuccessfull connection or retryable status codes
             bool retry = (statusCode < 0 || statusCode >= 429) && _writeOptions._maxRetryAttempts > 0;
             success = statusCode >= 200 && statusCode < 300;
@@ -490,7 +518,11 @@ bool InfluxDBClient::validateConnection() {
     }
     INFLUXDB_CLIENT_DEBUG("[D] Validating connection to %s\n", url.c_str());
 
-    return _service->doGET(url.c_str(), 200, nullptr);
+    bool ret = _service->doGET(url.c_str(), 200, nullptr);
+    if(!ret) {
+        INFLUXDB_CLIENT_DEBUG("[D] error %d: %s\n", _service->getLastStatusCode(), _service->getLastErrorMessage().c_str());
+    }
+    return ret;
 }
 
 int InfluxDBClient::postData(const char *data) {
@@ -500,14 +532,35 @@ int InfluxDBClient::postData(const char *data) {
     if(data) {
         INFLUXDB_CLIENT_DEBUG("[D] Writing to %s\n", _writeUrl.c_str());
         INFLUXDB_CLIENT_DEBUG("[D] Sending:\n%s\n", data);       
-
-        _service->doPOST(_writeUrl.c_str(), data, PSTR("text/plain"), 204, nullptr);
+        if(!_service->doPOST(_writeUrl.c_str(), data, PSTR("text/plain"), 204, nullptr)) {
+            INFLUXDB_CLIENT_DEBUG("[D] error %d: %s\n", _service->getLastStatusCode(), _service->getLastErrorMessage().c_str());
+        }
         _retryTime = _service->getLastRetryAfter();
         return _service->getLastStatusCode();
     } 
     return 0;
 }
 
+int InfluxDBClient::postData(Batch *batch) {
+    if(!_service && !init()) {
+        return 0;
+    }
+
+    BatchStreamer *bs = new BatchStreamer(batch);
+    INFLUXDB_CLIENT_DEBUG("[D] Writing to %s\n", _writeUrl.c_str());
+    INFLUXDB_CLIENT_DEBUG("[D] Sending:\n");       
+    
+    if(!_service->doPOST(_writeUrl.c_str(), bs, PSTR("text/plain"), 204, nullptr)) {
+        INFLUXDB_CLIENT_DEBUG("[D] error %d: %s\n", _service->getLastStatusCode(), _service->getLastErrorMessage().c_str());
+    }
+    delete bs;
+    _retryTime = _service->getLastRetryAfter();
+    return _service->getLastStatusCode();
+}
+
+void InfluxDBClient::setStreamWrite(bool enable) {
+    _streamWrite = enable;
+}
 
 
 static const char QueryDialect[] PROGMEM = "\
@@ -598,3 +651,78 @@ static String escapeJSONString(String &value) {
     }
     return ret;
 }
+
+InfluxDBClient::BatchStreamer::BatchStreamer(InfluxDBClient::Batch *batch) {
+    _batch = batch;
+    _read = 0;
+    _length = 0;
+    _pointer = 0;
+    _linePointer = 0;
+    for(int i=0;i<_batch->pointer;i++) {
+    _length += strlen(_batch->buffer[i])+1;
+    }
+}
+
+int InfluxDBClient::BatchStreamer::available() {
+    return _length-_read;
+}
+
+int InfluxDBClient::BatchStreamer::availableForWrite() {
+    return 0;
+}
+
+#if defined(ESP8266)        
+int InfluxDBClient::BatchStreamer::read(uint8_t* buffer, size_t len) {
+    INFLUXDB_CLIENT_DEBUG("BatchStream::read %d\n", len);
+    return readBytes((char *)buffer, len);
+}
+#endif
+size_t InfluxDBClient::BatchStreamer::readBytes(char* buffer, size_t len) {
+
+    INFLUXDB_CLIENT_DEBUG("BatchStream::readBytes %d\n", len);
+    int r=0;
+    for(int i=0;i<len;i++) {
+        if(available()) {
+            buffer[i] = read();
+            r++;
+        } else {
+            break;
+        }
+    }
+    return r;
+}
+
+int InfluxDBClient::BatchStreamer::read()  {
+    int r = peek();
+    if(r > 0) {
+        ++_read;
+        ++_linePointer;
+        if(!_batch->buffer[_pointer][_linePointer-1]) {
+            ++_pointer;
+            _linePointer = 0;
+        }
+    }
+    return r;
+}
+
+int InfluxDBClient::BatchStreamer::peek() {
+    if(_pointer == _batch->pointer) {
+        //This should not happen
+        return -1;
+    }
+    
+    int r;
+    if(!_batch->buffer[_pointer][_linePointer]) {
+        r = '\n';
+    } else {
+        r = _batch->buffer[_pointer][_linePointer];
+    }
+    // #ifdef INFLUXDB_CLIENT_DEBUG_ENABLE
+    //     Serial.printf_P(PSTR("%c"), r);
+    // #endif
+    return r;
+}
+
+size_t InfluxDBClient::BatchStreamer::write(uint8_t data)  {
+    return 0;
+}