#158

Build # #158

Build Type

push

Committed by

Commit Message

<a href="https://github.com/apache/datasketches-java/commit/<a class=hub.com/apache/datasketches-java/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/datasketches-java/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/datasketches-java/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/datasketches-java/commit/11cddab59a2f4e03f5248069e3aac4d941449bfb">11cddab59">&lt;a href=&quot;https://github.com/apache/datasketches-java/commit/</a><a class="double-link" href="https://github.com/apache/datasketches-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/datasketches-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/datasketches-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/datasketches-java/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/datasketches-java/commit/11cddab59a2f4e03f5248069e3aac4d941449bfb">11cddab59</a><a href="https://github.com/apache/datasketches-java/commit/11cddab59a2f4e03f5248069e3aac4d941449bfb">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/datasketches-java/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/datasketches-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/datasketches-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/datasketches-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/datasketches-java/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/datasketches-java/commit/11cddab59a2f4e03f5248069e3aac4d941449bfb&quot;&gt;11cddab59&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/datasketches-java/commit/11cddab59a2f4e03f5248069e3aac4d941449bfb&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/datasketches-java/commit/&amp;lt;/a&amp;gt;&amp;lt;a... (continued)

Run Details

17921 of 18222 relevant lines covered (98.35%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.14

/src/main/java/org/apache/datasketches/theta/ConcurrentBackgroundThetaPropagation.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

package org.apache.datasketches.theta;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Background propagation thread. Propagates a given sketch or a hash value from local threads
 * buffers into the shared sketch which stores the most up-to-date estimation of number of unique
 * items. This propagation is done at the background by dedicated threads, which allows
 * application threads to continue updating their local buffer.
 *
 * @author eshcar
 */
class ConcurrentBackgroundThetaPropagation implements Runnable {

  // Shared sketch to absorb the data
  private final ConcurrentSharedThetaSketch sharedThetaSketch;

  // Propagation flag of local buffer that is being processed.
  // It is the synchronization primitive to coordinate the work of the propagation with the
  // local buffer.  Updated when the propagation completes.
  private final AtomicBoolean localPropagationInProgress;


  // Sketch to be propagated to shared sketch. Can be null if only a single hash is propagated
  private final Sketch sketchIn;

  // Hash of the datum to be propagated to shared sketch. Can be ConcurrentSharedThetaSketch.NOT_SINGLE_HASH
  // if the data is propagated through a sketch.
  private final long singleHash;


  // The propagation epoch. The data can be propagated only within the context of this epoch.
  // The data should not be propagated if this epoch is not equal to the
  // shared sketch epoch.
  private final long epoch;

  ConcurrentBackgroundThetaPropagation(final ConcurrentSharedThetaSketch sharedThetaSketch,
      final AtomicBoolean localPropagationInProgress, final Sketch sketchIn, final long singleHash,
      final long epoch) {
    this.sharedThetaSketch = sharedThetaSketch;
    this.localPropagationInProgress = localPropagationInProgress;
    this.sketchIn = sketchIn;
    this.singleHash = singleHash;
    this.epoch = epoch;
  }

  /**
   * Propagation protocol:
   * 1) validate propagation is executed at the context of the right epoch, otherwise abort
   * 2) handle propagation: either of a single hash or of a sketch
   * 3) complete propagation: ping local buffer
   */
  @Override
  public void run() {
    // 1) validate propagation is executed at the context of the right epoch, otherwise abort
    if (!sharedThetaSketch.validateEpoch(epoch)) {
      // invalid epoch - should not propagate
      sharedThetaSketch.endPropagation(null, false);
      return;
    }

    // 2) handle propagation: either of a single hash or of a sketch
    if (singleHash != ConcurrentSharedThetaSketch.NOT_SINGLE_HASH) {
      sharedThetaSketch.propagate(singleHash);
    } else if (sketchIn != null) {
      final long volTheta = sharedThetaSketch.getVolatileTheta();
      assert volTheta <= sketchIn.getThetaLong() :
          "volTheta = " + volTheta + ", bufTheta = " + sketchIn.getThetaLong();

      // propagate values from input sketch one by one
      final long[] cacheIn = sketchIn.getCache();

      if (sketchIn.isOrdered()) { //Ordered compact, Use early stop
        for (final long hashIn : cacheIn) {
          if (hashIn >= volTheta) {
            break; //early stop
          }
          sharedThetaSketch.propagate(hashIn);
        }
      } else { //not ordered, also may have zeros (gaps) in the array.
        for (final long hashIn : cacheIn) {
          if (hashIn > 0) {
            sharedThetaSketch.propagate(hashIn);
          }
        }
      }
    }

    // 3) complete propagation: ping local buffer
    sharedThetaSketch.endPropagation(localPropagationInProgress, false);
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.datasketches.theta;
21
22	import java.util.concurrent.atomic.AtomicBoolean;
23
24	/**
25	* Background propagation thread. Propagates a given sketch or a hash value from local threads
26	* buffers into the shared sketch which stores the most up-to-date estimation of number of unique
27	* items. This propagation is done at the background by dedicated threads, which allows
28	* application threads to continue updating their local buffer.
29	*
30	* @author eshcar
31	*/
32	class ConcurrentBackgroundThetaPropagation implements Runnable {	1✔
33
34	// Shared sketch to absorb the data
35	private final ConcurrentSharedThetaSketch sharedThetaSketch;
36
37	// Propagation flag of local buffer that is being processed.
38	// It is the synchronization primitive to coordinate the work of the propagation with the
39	// local buffer. Updated when the propagation completes.
40	private final AtomicBoolean localPropagationInProgress;
41
42
43	// Sketch to be propagated to shared sketch. Can be null if only a single hash is propagated
44	private final Sketch sketchIn;
45
46	// Hash of the datum to be propagated to shared sketch. Can be ConcurrentSharedThetaSketch.NOT_SINGLE_HASH
47	// if the data is propagated through a sketch.
48	private final long singleHash;
49
50
51	// The propagation epoch. The data can be propagated only within the context of this epoch.
52	// The data should not be propagated if this epoch is not equal to the
53	// shared sketch epoch.
54	private final long epoch;
55
56	ConcurrentBackgroundThetaPropagation(final ConcurrentSharedThetaSketch sharedThetaSketch,
57	final AtomicBoolean localPropagationInProgress, final Sketch sketchIn, final long singleHash,
58	final long epoch) {	1✔
59	this.sharedThetaSketch = sharedThetaSketch;	1✔
60	this.localPropagationInProgress = localPropagationInProgress;	1✔
61	this.sketchIn = sketchIn;	1✔
62	this.singleHash = singleHash;	1✔
63	this.epoch = epoch;	1✔
64	}	1✔
65
66	/**
67	* Propagation protocol:
68	* 1) validate propagation is executed at the context of the right epoch, otherwise abort
69	* 2) handle propagation: either of a single hash or of a sketch
70	* 3) complete propagation: ping local buffer
71	*/
72	@Override
73	public void run() {
74	// 1) validate propagation is executed at the context of the right epoch, otherwise abort
75	if (!sharedThetaSketch.validateEpoch(epoch)) {	1✔
76	// invalid epoch - should not propagate
77	sharedThetaSketch.endPropagation(null, false);	×
78	return;	×
79	}
80
81	// 2) handle propagation: either of a single hash or of a sketch
82	if (singleHash != ConcurrentSharedThetaSketch.NOT_SINGLE_HASH) {	1✔
83	sharedThetaSketch.propagate(singleHash);	×
84	} else if (sketchIn != null) {	1✔
85	final long volTheta = sharedThetaSketch.getVolatileTheta();	1✔
86	assert volTheta <= sketchIn.getThetaLong() :	1✔
87	"volTheta = " + volTheta + ", bufTheta = " + sketchIn.getThetaLong();	×
88
89	// propagate values from input sketch one by one
90	final long[] cacheIn = sketchIn.getCache();	1✔
91
92	if (sketchIn.isOrdered()) { //Ordered compact, Use early stop	1✔
93	for (final long hashIn : cacheIn) {	1✔
94	if (hashIn >= volTheta) {	1✔
95	break; //early stop	×
96	}
97	sharedThetaSketch.propagate(hashIn);	1✔
98	}
99	} else { //not ordered, also may have zeros (gaps) in the array.
100	for (final long hashIn : cacheIn) {	1✔
101	if (hashIn > 0) {	1✔
102	sharedThetaSketch.propagate(hashIn);	1✔
103	}
104	}
105	}
106	}
107
108	// 3) complete propagation: ping local buffer
109	sharedThetaSketch.endPropagation(localPropagationInProgress, false);	1✔
110	}	1✔
111
112	}

apache / datasketches-java / #158

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous