#10019

Committed 07 Sep 2023 04:50AM UTC coverage: 47.489% (-0.2%) from 47.655%

Build # #10019

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

Pipe: Fix ConcurrentModificationException caused by concurrently iterating through CachedSchemaPatternMatcher.extractors when an PipeHeartbeatEvent is being assigned (#11074)

* try to fix ConcurrentModificationException when assigning PipeHeartbeatEvent

* Update CachedSchemaPatternMatcher.java

---------

Co-authored-by: 马子坤 <55695098+DanielWang2035@users.noreply.github.com>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

80551 of 169622 relevant lines covered (47.49%)

0.47 hits per line

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

0.0

/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/model/ModelInformation.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.iotdb.commons.model;

import org.apache.iotdb.common.rpc.thrift.TaskType;
import org.apache.iotdb.common.rpc.thrift.TrainingState;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;

import javax.annotation.Nullable;

import java.io.DataOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.apache.iotdb.commons.model.TrialInformation.MODEL_PATH;

public abstract class ModelInformation {

  private final String modelId;

  private final Map<String, String> options;
  private final String datasetFetchSQL;

  private TrainingState trainingState;

  @Nullable private String bestTrialId;
  private final Map<String, TrialInformation> trialMap;

  public static final String TASK_TYPE = "task_type";
  public static final String MODEL_TYPE = "model_type";

  protected ModelInformation(String modelId, Map<String, String> options, String datasetFetchSql) {
    this.modelId = modelId;
    this.options = options;
    this.datasetFetchSQL = datasetFetchSql;
    this.trainingState = TrainingState.PENDING;
    this.trialMap = new HashMap<>();
  }

  protected ModelInformation(ByteBuffer buffer) {
    this.modelId = ReadWriteIOUtils.readString(buffer);

    this.options = ReadWriteIOUtils.readMap(buffer);
    this.datasetFetchSQL = ReadWriteIOUtils.readString(buffer);

    this.trainingState = TrainingState.findByValue(ReadWriteIOUtils.readInt(buffer));

    byte isNull = ReadWriteIOUtils.readByte(buffer);
    if (isNull == 1) {
      this.bestTrialId = ReadWriteIOUtils.readString(buffer);
    }

    int mapSize = ReadWriteIOUtils.readInt(buffer);
    this.trialMap = new HashMap<>();
    for (int i = 0; i < mapSize; i++) {
      TrialInformation trialInformation = TrialInformation.deserialize(buffer);
      this.trialMap.put(trialInformation.getTrialId(), trialInformation);
    }
  }

  protected ModelInformation(InputStream stream) throws IOException {
    this.modelId = ReadWriteIOUtils.readString(stream);

    this.options = ReadWriteIOUtils.readMap(stream);
    this.datasetFetchSQL = ReadWriteIOUtils.readString(stream);

    this.trainingState = TrainingState.findByValue(ReadWriteIOUtils.readInt(stream));

    byte isNull = ReadWriteIOUtils.readByte(stream);
    if (isNull == 1) {
      this.bestTrialId = ReadWriteIOUtils.readString(stream);
    }

    int mapSize = ReadWriteIOUtils.readInt(stream);
    this.trialMap = new HashMap<>();
    for (int i = 0; i < mapSize; i++) {
      TrialInformation trialInformation = TrialInformation.deserialize(stream);
      this.trialMap.put(trialInformation.getTrialId(), trialInformation);
    }
  }

  public abstract TaskType getTaskType();

  public String getModelId() {
    return modelId;
  }

  private String getModelType() {
    return options.get(MODEL_TYPE);
  }

  public Map<String, String> getOptions() {
    return options;
  }

  public String getDatasetFetchSql() {
    return datasetFetchSQL;
  }

  public boolean available() {
    return trainingState == TrainingState.FINISHED;
  }

  public TrialInformation getTrialInformationById(String trialId) {
    if (trialMap.containsKey(trialId)) {
      return trialMap.get(trialId);
    }
    return null;
  }

  public List<TrialInformation> getAllTrialInformation() {
    return new ArrayList<>(trialMap.values());
  }

  public void update(String trailId, Map<String, String> modelInfo) {
    if (!trialMap.containsKey(trailId)) {
      String modelPath = null;
      if (modelInfo.containsKey(MODEL_PATH)) {
        modelPath = modelInfo.get(MODEL_PATH);
        modelInfo.remove(MODEL_PATH);
      }
      TrialInformation trialInformation =
          new TrialInformation(trailId, new ModelHyperparameter(modelInfo), modelPath);
      trialMap.put(trailId, trialInformation);
    } else {
      trialMap.get(trailId).update(modelInfo);
    }
  }

  public void updateState(TrainingState newState, String bestTrailId) {
    // TODO: add state transform validate
    this.trainingState = newState;
    if (bestTrailId != null) {
      this.bestTrialId = bestTrailId;
    }
  }

  public String getModelPath() {
    if (bestTrialId != null) {
      TrialInformation bestTrail = trialMap.get(bestTrialId);
      return bestTrail.getModelPath();
    } else {
      return "UNKNOWN";
    }
  }

  public void serialize(DataOutputStream stream) throws IOException {
    ReadWriteIOUtils.write(modelId, stream);

    ReadWriteIOUtils.write(options, stream);
    ReadWriteIOUtils.write(datasetFetchSQL, stream);

    ReadWriteIOUtils.write(trainingState.ordinal(), stream);

    if (bestTrialId == null) {
      ReadWriteIOUtils.write((byte) 0, stream);
    } else {
      ReadWriteIOUtils.write((byte) 1, stream);
      ReadWriteIOUtils.write(bestTrialId, stream);
    }

    ReadWriteIOUtils.write(trialMap.size(), stream);
    for (TrialInformation trialInformation : trialMap.values()) {
      trialInformation.serialize(stream);
    }
  }

  public void serialize(FileOutputStream stream) throws IOException {
    ReadWriteIOUtils.write(modelId, stream);

    ReadWriteIOUtils.write(options, stream);
    ReadWriteIOUtils.write(datasetFetchSQL, stream);

    ReadWriteIOUtils.write(trainingState.ordinal(), stream);

    if (bestTrialId == null) {
      ReadWriteIOUtils.write((byte) 0, stream);
    } else {
      ReadWriteIOUtils.write((byte) 1, stream);
      ReadWriteIOUtils.write(bestTrialId, stream);
    }

    ReadWriteIOUtils.write(trialMap.size(), stream);
    for (TrialInformation trialInformation : trialMap.values()) {
      trialInformation.serialize(stream);
    }
  }

  public static ModelInformation deserialize(ByteBuffer buffer) {
    TaskType modelTask = TaskType.findByValue(ReadWriteIOUtils.readInt(buffer));
    if (modelTask == null) {
      throw new IllegalArgumentException();
    }

    if (modelTask == TaskType.FORECAST) {
      return new ForecastModeInformation(buffer);
    }
    throw new IllegalArgumentException("Invalid task type: " + modelTask);
  }

  public static ModelInformation deserialize(InputStream stream) throws IOException {
    TaskType modelTask = TaskType.findByValue(ReadWriteIOUtils.readInt(stream));
    if (modelTask == null) {
      throw new IllegalArgumentException();
    }

    if (modelTask == TaskType.FORECAST) {
      return new ForecastModeInformation(stream);
    }
    throw new IllegalArgumentException("Invalid task type: " + modelTask);
  }

  public ByteBuffer serializeShowModelResult() throws IOException {
    PublicBAOS buffer = new PublicBAOS();
    DataOutputStream stream = new DataOutputStream(buffer);
    ReadWriteIOUtils.write(modelId, stream);
    ReadWriteIOUtils.write(getTaskType().toString(), stream);
    ReadWriteIOUtils.write(getModelType(), stream);
    ReadWriteIOUtils.write(datasetFetchSQL, stream);
    ReadWriteIOUtils.write(trainingState.toString(), stream);

    if (bestTrialId != null) {
      TrialInformation bestTrail = trialMap.get(bestTrialId);
      List<String> modelHyperparameterList = bestTrail.getModelHyperparameter().toStringList();
      ReadWriteIOUtils.write(modelHyperparameterList.size() + 1, stream);
      for (String hyperparameter : modelHyperparameterList) {
        ReadWriteIOUtils.write(hyperparameter, stream);
      }
    } else {
      ReadWriteIOUtils.write(2, stream);
      ReadWriteIOUtils.write("UNKNOWN", stream);
    }
    // add extra blank line to make the result more readable in cli
    ReadWriteIOUtils.write(" ", stream);
    return ByteBuffer.wrap(buffer.getBuf(), 0, buffer.size());
  }
}

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.iotdb.commons.model;
21
22	import org.apache.iotdb.common.rpc.thrift.TaskType;
23	import org.apache.iotdb.common.rpc.thrift.TrainingState;
24	import org.apache.iotdb.tsfile.utils.PublicBAOS;
25	import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
26
27	import javax.annotation.Nullable;
28
29	import java.io.DataOutputStream;
30	import java.io.FileOutputStream;
31	import java.io.IOException;
32	import java.io.InputStream;
33	import java.nio.ByteBuffer;
34	import java.util.ArrayList;
35	import java.util.HashMap;
36	import java.util.List;
37	import java.util.Map;
38
39	import static org.apache.iotdb.commons.model.TrialInformation.MODEL_PATH;
40
41	public abstract class ModelInformation {
42
43	private final String modelId;
44
45	private final Map<String, String> options;
46	private final String datasetFetchSQL;
47
48	private TrainingState trainingState;
49
50	@Nullable private String bestTrialId;
51	private final Map<String, TrialInformation> trialMap;
52
53	public static final String TASK_TYPE = "task_type";
54	public static final String MODEL_TYPE = "model_type";
55
56	protected ModelInformation(String modelId, Map<String, String> options, String datasetFetchSql) {	×
57	this.modelId = modelId;	×
58	this.options = options;	×
59	this.datasetFetchSQL = datasetFetchSql;	×
60	this.trainingState = TrainingState.PENDING;	×
61	this.trialMap = new HashMap<>();	×
62	}	×
63
64	protected ModelInformation(ByteBuffer buffer) {	×
65	this.modelId = ReadWriteIOUtils.readString(buffer);	×
66
67	this.options = ReadWriteIOUtils.readMap(buffer);	×
68	this.datasetFetchSQL = ReadWriteIOUtils.readString(buffer);	×
69
70	this.trainingState = TrainingState.findByValue(ReadWriteIOUtils.readInt(buffer));	×
71
72	byte isNull = ReadWriteIOUtils.readByte(buffer);	×
73	if (isNull == 1) {	×
74	this.bestTrialId = ReadWriteIOUtils.readString(buffer);	×
75	}
76
77	int mapSize = ReadWriteIOUtils.readInt(buffer);	×
78	this.trialMap = new HashMap<>();	×
79	for (int i = 0; i < mapSize; i++) {	×
80	TrialInformation trialInformation = TrialInformation.deserialize(buffer);	×
81	this.trialMap.put(trialInformation.getTrialId(), trialInformation);	×
82	}
83	}	×
84
85	protected ModelInformation(InputStream stream) throws IOException {	×
86	this.modelId = ReadWriteIOUtils.readString(stream);	×
87
88	this.options = ReadWriteIOUtils.readMap(stream);	×
89	this.datasetFetchSQL = ReadWriteIOUtils.readString(stream);	×
90
91	this.trainingState = TrainingState.findByValue(ReadWriteIOUtils.readInt(stream));	×
92
93	byte isNull = ReadWriteIOUtils.readByte(stream);	×
94	if (isNull == 1) {	×
95	this.bestTrialId = ReadWriteIOUtils.readString(stream);	×
96	}
97
98	int mapSize = ReadWriteIOUtils.readInt(stream);	×
99	this.trialMap = new HashMap<>();	×
100	for (int i = 0; i < mapSize; i++) {	×
101	TrialInformation trialInformation = TrialInformation.deserialize(stream);	×
102	this.trialMap.put(trialInformation.getTrialId(), trialInformation);	×
103	}
104	}	×
105
106	public abstract TaskType getTaskType();
107
108	public String getModelId() {
109	return modelId;	×
110	}
111
112	private String getModelType() {
113	return options.get(MODEL_TYPE);	×
114	}
115
116	public Map<String, String> getOptions() {
117	return options;	×
118	}
119
120	public String getDatasetFetchSql() {
121	return datasetFetchSQL;	×
122	}
123
124	public boolean available() {
125	return trainingState == TrainingState.FINISHED;	×
126	}
127
128	public TrialInformation getTrialInformationById(String trialId) {
129	if (trialMap.containsKey(trialId)) {	×
130	return trialMap.get(trialId);	×
131	}
132	return null;	×
133	}
134
135	public List<TrialInformation> getAllTrialInformation() {
136	return new ArrayList<>(trialMap.values());	×
137	}
138
139	public void update(String trailId, Map<String, String> modelInfo) {
140	if (!trialMap.containsKey(trailId)) {	×
141	String modelPath = null;	×
142	if (modelInfo.containsKey(MODEL_PATH)) {	×
143	modelPath = modelInfo.get(MODEL_PATH);	×
144	modelInfo.remove(MODEL_PATH);	×
145	}
146	TrialInformation trialInformation =	×
147	new TrialInformation(trailId, new ModelHyperparameter(modelInfo), modelPath);
148	trialMap.put(trailId, trialInformation);	×
149	} else {	×
150	trialMap.get(trailId).update(modelInfo);	×
151	}
152	}	×
153
154	public void updateState(TrainingState newState, String bestTrailId) {
155	// TODO: add state transform validate
156	this.trainingState = newState;	×
157	if (bestTrailId != null) {	×
158	this.bestTrialId = bestTrailId;	×
159	}
160	}	×
161
162	public String getModelPath() {
163	if (bestTrialId != null) {	×
164	TrialInformation bestTrail = trialMap.get(bestTrialId);	×
165	return bestTrail.getModelPath();	×
166	} else {
167	return "UNKNOWN";	×
168	}
169	}
170
171	public void serialize(DataOutputStream stream) throws IOException {
172	ReadWriteIOUtils.write(modelId, stream);	×
173
174	ReadWriteIOUtils.write(options, stream);	×
175	ReadWriteIOUtils.write(datasetFetchSQL, stream);	×
176
177	ReadWriteIOUtils.write(trainingState.ordinal(), stream);	×
178
179	if (bestTrialId == null) {	×
180	ReadWriteIOUtils.write((byte) 0, stream);	×
181	} else {
182	ReadWriteIOUtils.write((byte) 1, stream);	×
183	ReadWriteIOUtils.write(bestTrialId, stream);	×
184	}
185
186	ReadWriteIOUtils.write(trialMap.size(), stream);	×
187	for (TrialInformation trialInformation : trialMap.values()) {	×
188	trialInformation.serialize(stream);	×
189	}	×
190	}	×
191
192	public void serialize(FileOutputStream stream) throws IOException {
193	ReadWriteIOUtils.write(modelId, stream);	×
194
195	ReadWriteIOUtils.write(options, stream);	×
196	ReadWriteIOUtils.write(datasetFetchSQL, stream);	×
197
198	ReadWriteIOUtils.write(trainingState.ordinal(), stream);	×
199
200	if (bestTrialId == null) {	×
201	ReadWriteIOUtils.write((byte) 0, stream);	×
202	} else {
203	ReadWriteIOUtils.write((byte) 1, stream);	×
204	ReadWriteIOUtils.write(bestTrialId, stream);	×
205	}
206
207	ReadWriteIOUtils.write(trialMap.size(), stream);	×
208	for (TrialInformation trialInformation : trialMap.values()) {	×
209	trialInformation.serialize(stream);	×
210	}	×
211	}	×
212
213	public static ModelInformation deserialize(ByteBuffer buffer) {
214	TaskType modelTask = TaskType.findByValue(ReadWriteIOUtils.readInt(buffer));	×
215	if (modelTask == null) {	×
216	throw new IllegalArgumentException();	×
217	}
218
219	if (modelTask == TaskType.FORECAST) {	×
220	return new ForecastModeInformation(buffer);	×
221	}
222	throw new IllegalArgumentException("Invalid task type: " + modelTask);	×
223	}
224
225	public static ModelInformation deserialize(InputStream stream) throws IOException {
226	TaskType modelTask = TaskType.findByValue(ReadWriteIOUtils.readInt(stream));	×
227	if (modelTask == null) {	×
228	throw new IllegalArgumentException();	×
229	}
230
231	if (modelTask == TaskType.FORECAST) {	×
232	return new ForecastModeInformation(stream);	×
233	}
234	throw new IllegalArgumentException("Invalid task type: " + modelTask);	×
235	}
236
237	public ByteBuffer serializeShowModelResult() throws IOException {
238	PublicBAOS buffer = new PublicBAOS();	×
239	DataOutputStream stream = new DataOutputStream(buffer);	×
240	ReadWriteIOUtils.write(modelId, stream);	×
241	ReadWriteIOUtils.write(getTaskType().toString(), stream);	×
242	ReadWriteIOUtils.write(getModelType(), stream);	×
243	ReadWriteIOUtils.write(datasetFetchSQL, stream);	×
244	ReadWriteIOUtils.write(trainingState.toString(), stream);	×
245
246	if (bestTrialId != null) {	×
247	TrialInformation bestTrail = trialMap.get(bestTrialId);	×
248	List<String> modelHyperparameterList = bestTrail.getModelHyperparameter().toStringList();	×
249	ReadWriteIOUtils.write(modelHyperparameterList.size() + 1, stream);	×
250	for (String hyperparameter : modelHyperparameterList) {	×
251	ReadWriteIOUtils.write(hyperparameter, stream);	×
252	}	×
253	} else {	×
254	ReadWriteIOUtils.write(2, stream);	×
255	ReadWriteIOUtils.write("UNKNOWN", stream);	×
256	}
257	// add extra blank line to make the result more readable in cli
258	ReadWriteIOUtils.write(" ", stream);	×
259	return ByteBuffer.wrap(buffer.getBuf(), 0, buffer.size());	×
260	}
261	}

apache / iotdb / #10019

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