Skip to content

Commit 59b93a0

Browse files
pasinpasin
authored
Merge pull request #1577 from couchbase/issue/1569
Fixed #1569 - Out Of MemoryError on replication
2 parents dd087d2 + 9752afb commit 59b93a0

6 files changed

Lines changed: 75 additions & 10 deletions

File tree

‎src/main/java/com/couchbase/lite/internal/Body.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
*/
3232
public class Body {
3333
private byte[] json;
34+
private long size = 0;
3435
private Object object;
3536

3637
public Body(byte[] json) {
@@ -41,6 +42,16 @@ public Body(Map<String, Object> properties) {
4142
this.object = properties;
4243
}
4344

45+
public Body(byte[] json, long size) {
46+
this.json = json;
47+
this.size = size;
48+
}
49+
50+
public Body(Map<String, Object> properties, long size) {
51+
this.object = properties;
52+
this.size = size;
53+
}
54+
4455
public Body(List<?> array) {
4556
this.object = array;
4657
}
@@ -171,4 +182,8 @@ public void release() {
171182
public Object getObject(String key) {
172183
return getProperties() != null ? getProperties().get(key) : null;
173184
}
185+
186+
public long getSize() {
187+
return size;
188+
}
174189
}

‎src/main/java/com/couchbase/lite/internal/RevisionInternal.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,10 @@ public RevisionInternal(Map<String, Object> properties) {
6464
this(new Body(properties));
6565
}
6666

67+
public RevisionInternal(Map<String, Object> properties, long size) {
68+
this(new Body(properties, size));
69+
}
70+
6771
////////////////////////////////////////////////////////////
6872
// Setter / Getter methods
6973
////////////////////////////////////////////////////////////

‎src/main/java/com/couchbase/lite/replicator/PulledRevision.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ public PulledRevision(Map<String, Object> properties) {
4040
super(properties);
4141
}
4242

43+
public PulledRevision(Map<String, Object> properties, long size) {
44+
super(properties, size);
45+
}
46+
4347
public String getRemoteSequenceID() {
4448
return remoteSequenceID;
4549
}

‎src/main/java/com/couchbase/lite/replicator/PullerInternal.java‎

Lines changed: 42 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import java.util.concurrent.Future;
4848
import java.util.concurrent.TimeUnit;
4949
import java.util.concurrent.atomic.AtomicBoolean;
50+
import java.util.concurrent.atomic.AtomicLong;
5051

5152
import okhttp3.OkHttpClient;
5253
import okhttp3.Response;
@@ -77,6 +78,8 @@ public class PullerInternal extends ReplicationInternal implements ChangeTracker
7778
private static final int INSERTION_BATCHER_DELAY = 250; // 0.25 Seconds
7879
private static final int INSERTION_BATCHER_CAPACITY = 100;
7980

81+
private static final long MAX_QUEUE_MEMORY_SIZE = 2 * 1024 * 1024; // 2MB
82+
8083
private ChangeTracker changeTracker;
8184
protected SequenceMap pendingSequences;
8285
protected Boolean canBulkGet; // Does the server support _bulk_get requests?
@@ -88,6 +91,7 @@ public class PullerInternal extends ReplicationInternal implements ChangeTracker
8891
new ArrayList<RevisionInternal>(100));
8992
protected int httpConnectionCount;
9093
protected Batcher<RevisionInternal> downloadsToInsert;
94+
protected AtomicLong queuedMemorySize = new AtomicLong(0);
9195

9296
private String str = null;
9397

@@ -125,6 +129,9 @@ private void initDownloadsToInsert() {
125129
@Override
126130
public void process(List<RevisionInternal> inbox) {
127131
insertDownloads(inbox);
132+
if (downloadsToInsert.count() == 0) {
133+
queuedMemorySize.set(0);
134+
}
128135
}
129136
});
130137
}
@@ -332,12 +339,12 @@ protected void pullBulkRevisions(List<RevisionInternal> bulkRevs) {
332339
db,
333340
this.requestHeaders,
334341
new RemoteBulkDownloaderRequest.BulkDownloaderDocument() {
335-
public void onDocument(Map<String, Object> props) {
342+
public void onDocument(Map<String, Object> props, long size) {
336343
// Got a revision!
337344
// Find the matching revision in 'remainingRevs' and get its sequence:
338345
RevisionInternal rev;
339346
if (props.get("_id") != null) {
340-
rev = new RevisionInternal(props);
347+
rev = new RevisionInternal(props, size);
341348
} else {
342349
rev = new RevisionInternal((String) props.get("id"),
343350
(String) props.get("rev"), false);
@@ -456,12 +463,22 @@ private void queueDownloadedRevision(RevisionInternal rev) {
456463
}
457464
}
458465

459-
if (rev != null && rev.getBody() != null)
460-
rev.getBody().compact();
466+
// NOTE: should not/not necessary to call Body.compact()
467+
// new RevisionInternal(Map<string, Object>) creates Body instance only
468+
// with `object`. Serializing object to json causes two unnecessary
469+
// JSON serializations.
470+
471+
if (rev.getBody() != null)
472+
queuedMemorySize.addAndGet(rev.getBody().getSize());
461473

462474
downloadsToInsert.queueObject(rev);
463-
}
464475

476+
// if queue memory size is more than maximum, force flush the queue.
477+
if (queuedMemorySize.get() > MAX_QUEUE_MEMORY_SIZE) {
478+
Log.d(TAG, "Flushing queued memory size at: " + queuedMemorySize);
479+
downloadsToInsert.flushAll(true);
480+
}
481+
}
465482

466483
// Get as many revisions as possible in one _all_docs request.
467484
// This is compatible with CouchDB, but it only works for revs of generation 1 without attachments.
@@ -586,7 +603,9 @@ public boolean run() {
586603
continue;
587604
}
588605
}
589-
if (rev.getBody() != null) rev.getBody().compact();
606+
607+
// NOTE: calling Body.compact() here cause another JSON serialization by Jackson.
608+
// At this point Body.json is null, and Body.object has values.
590609

591610
// Mark this revision's fake sequence as processed:
592611
pendingSequences.removeSequence(fakeSequence);
@@ -700,18 +719,33 @@ public void onCompletion(Response httpResponse, Object result, Throwable e) {
700719
setError(e);
701720
}
702721
} else {
722+
703723
Map<String, Object> properties = (Map<String, Object>) result;
704-
PulledRevision gotRev = new PulledRevision(properties);
724+
long size = 0;
725+
if (httpResponse != null && httpResponse.body() != null)
726+
size = httpResponse.body().contentLength();
727+
PulledRevision gotRev = new PulledRevision(properties, size);
705728
gotRev.setSequence(rev.getSequence());
706729

707730
Log.d(TAG, "%s: pullRemoteRevision add rev: %s to batcher: %s",
708731
PullerInternal.this, gotRev, downloadsToInsert);
709732

733+
// NOTE: should not/not necessary to call Body.compact()
734+
// new PulledRevision(Map<string, Object>) creates Body instance only
735+
// with `object`. Serializing object to json causes two unnecessary
736+
// JSON serializations.
737+
710738
if (gotRev.getBody() != null)
711-
gotRev.getBody().compact();
739+
queuedMemorySize.addAndGet(gotRev.getBody().getSize());
712740

713741
// Add to batcher ... eventually it will be fed to -insertRevisions:.
714742
downloadsToInsert.queueObject(gotRev);
743+
744+
// if queue memory size is more than maximum, force flush the queue.
745+
if (queuedMemorySize.get() > MAX_QUEUE_MEMORY_SIZE) {
746+
Log.d(TAG, "Flushing queued memory size at: " + queuedMemorySize);
747+
downloadsToInsert.flushAll(true);
748+
}
715749
}
716750

717751
// Note that we've finished this task:

‎src/main/java/com/couchbase/lite/replicator/RemoteBulkDownloaderRequest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public class RemoteBulkDownloaderRequest extends RemoteRequest implements Multip
4949

5050
@InterfaceAudience.Private
5151
public interface BulkDownloaderDocument {
52-
void onDocument(Map<String, Object> props);
52+
void onDocument(Map<String, Object> props, long size);
5353
}
5454

5555
////////////////////////////////////////////////////////////
@@ -226,7 +226,7 @@ public void finishedPart() {
226226
if (_docReader == null)
227227
throw new IllegalStateException("_docReader is not defined");
228228
_docReader.finish();
229-
_onDocument.onDocument(_docReader.getDocumentProperties());
229+
_onDocument.onDocument(_docReader.getDocumentProperties(), _docReader.getDocumentSize());
230230
_docReader = null;
231231
Log.v(TAG, "%s: Finished document", this);
232232
}

‎src/main/java/com/couchbase/lite/support/MultipartDocumentReader.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ public class MultipartDocumentReader implements MultipartReaderDelegate {
3333
private CustomByteArrayOutputStream jsonBuffer;
3434
private boolean jsonCompressed;
3535
private Map<String, Object> document;
36+
private long documentSize = 0; // original JSON document size
3637
private Database database;
3738
private Map<String, BlobStoreWriter> attachmentsByName;
3839
private Map<String, BlobStoreWriter> attachmentsByMd5Digest;
@@ -45,7 +46,14 @@ public Map<String, Object> getDocumentProperties() {
4546
return document;
4647
}
4748

49+
public long getDocumentSize() {
50+
return documentSize;
51+
}
52+
4853
public void parseJsonBuffer() {
54+
55+
documentSize = jsonBuffer.count();
56+
4957
ByteArrayInputStream in = new ByteArrayInputStream(jsonBuffer.buf(), 0, jsonBuffer.count());
5058
try {
5159
try {

0 commit comments

Comments
 (0)