diff --git a/mindspore/ccsrc/minddata/dataset/util/status.cc b/mindspore/ccsrc/minddata/dataset/util/status.cc index 1a95170ae0b..f69166b03e0 100644 --- a/mindspore/ccsrc/minddata/dataset/util/status.cc +++ b/mindspore/ccsrc/minddata/dataset/util/status.cc @@ -15,9 +15,9 @@ */ #include "minddata/dataset/util/status.h" +#include +#include #include -#include -#include #include "./securec.h" #ifndef ENABLE_ANDROID @@ -32,8 +32,7 @@ namespace dataset { float GetMemoryUsage() { char buf[128] = {0}; - FILE *fd; - fd = fopen("/proc/meminfo", "r"); + FILE *fd = fopen("/proc/meminfo", "r"); if (fd == nullptr) { MS_LOG(WARNING) << "The meminfo file: /proc/meminfo is opened failed."; return 0.0; diff --git a/mindspore/ccsrc/minddata/dataset/util/task.cc b/mindspore/ccsrc/minddata/dataset/util/task.cc index 4d7b65cd408..682d6cb8f2c 100644 --- a/mindspore/ccsrc/minddata/dataset/util/task.cc +++ b/mindspore/ccsrc/minddata/dataset/util/task.cc @@ -172,7 +172,7 @@ Status Task::Join(WaitFlag blocking) { // just wait 30 seconds // case1: cpu usage 100%, DeviceQueueOp thread may destroy without thrd_ future - if (wait_times > 30) { + if (wait_times > kWaitInterruptTaskTime) { MS_LOG(WARNING) << MyName() << " Thread ID " << ss.str() << " is not responding. Maybe it's destroyed, task stop."; break; diff --git a/mindspore/ccsrc/minddata/dataset/util/task.h b/mindspore/ccsrc/minddata/dataset/util/task.h index 72294d1bd47..1917ece7268 100644 --- a/mindspore/ccsrc/minddata/dataset/util/task.h +++ b/mindspore/ccsrc/minddata/dataset/util/task.h @@ -40,6 +40,8 @@ namespace mindspore { namespace dataset { +const uint32_t kWaitInterruptTaskTime = 30; // the wait time of interrupt task + class TaskManager; class Task : public IntrpResource { diff --git a/mindspore/dataset/engine/cache_admin.py b/mindspore/dataset/engine/cache_admin.py index b2b8aa51af5..95a006636c3 100644 --- a/mindspore/dataset/engine/cache_admin.py +++ b/mindspore/dataset/engine/cache_admin.py @@ -23,6 +23,7 @@ import sys import mindspore def main(): + """Entry point for cache service""" cache_admin_dir = os.path.join(os.path.dirname(mindspore.__file__), "bin") os.chdir(cache_admin_dir) diff --git a/mindspore/dataset/engine/iterators.py b/mindspore/dataset/engine/iterators.py index f5499c2267c..71d0ad31dd9 100644 --- a/mindspore/dataset/engine/iterators.py +++ b/mindspore/dataset/engine/iterators.py @@ -84,7 +84,7 @@ class Iterator: self._transform_tensor = lambda t: Tensor(t.as_array()) else: self._transform_tensor = lambda t: Tensor.from_numpy(t.as_array()) - self._index = 0 + self.__index = 0 # todo remove next when ContextManager is done ITERATORS_LIST.append(weakref.ref(self)) @@ -123,12 +123,12 @@ class Iterator: data = self._get_next() if not data: - if self._index == 0: + if self.__index == 0: logger.warning("No records available.") if self._ori_dataset.dataset_size is None: - self._ori_dataset.dataset_size = self._index + self._ori_dataset.dataset_size = self.__index raise StopIteration - self._index += 1 + self.__index += 1 return data def __deepcopy__(self, memo): diff --git a/mindspore/mindrecord/tools/cifar100_to_mr.py b/mindspore/mindrecord/tools/cifar100_to_mr.py index 63ddecf08ed..f8192cbb1ca 100644 --- a/mindspore/mindrecord/tools/cifar100_to_mr.py +++ b/mindspore/mindrecord/tools/cifar100_to_mr.py @@ -24,7 +24,7 @@ from mindspore import log as logger from .cifar100 import Cifar100 from ..common.exceptions import PathNotExistsError from ..filewriter import FileWriter -from ..shardutils import check_filename, ExceptionThread, SUCCESS +from ..shardutils import check_filename, ExceptionThread, SUCCESS, FAILED try: cv2 = import_module("cv2") diff --git a/mindspore/mindrecord/tools/tfrecord_to_mr.py b/mindspore/mindrecord/tools/tfrecord_to_mr.py index 6d793e19543..5983097a4e8 100644 --- a/mindspore/mindrecord/tools/tfrecord_to_mr.py +++ b/mindspore/mindrecord/tools/tfrecord_to_mr.py @@ -124,7 +124,27 @@ class TFRecordToMR: self.bytes_fields_list = bytes_fields_list self.scalar_set = set() self.list_set = set() + self.mindrecord_schema = self._parse_mindrecord_schema_from_feature_dict() + def _check_input(self, source, destination, feature_dict): + """Validation check for inputs of init method""" + if not isinstance(source, str): + raise ValueError("Parameter source must be string.") + check_filename(source, "source") + + if not isinstance(destination, str): + raise ValueError("Parameter destination must be string.") + check_filename(destination, "destination") + + if feature_dict is None or not isinstance(feature_dict, dict): + raise ValueError("Parameter feature_dict is None or not dict.") + + for _, val in feature_dict.items(): + if not isinstance(val, self.tf.io.FixedLenFeature): + raise ValueError("Parameter feature_dict: {} only support FixedLenFeature.".format(feature_dict)) + + def _parse_mindrecord_schema_from_feature_dict(self): + """get mindrecord schema from feature dict""" mindrecord_schema = {} for key, val in self.feature_dict.items(): if not val.shape: @@ -146,24 +166,7 @@ class TFRecordToMR: "is not None. It is not supported.".format(key)) self.list_set.add(_cast_name(key)) mindrecord_schema[_cast_name(key)] = {"type": self._cast_type(val.dtype), "shape": [val.shape[0]]} - self.mindrecord_schema = mindrecord_schema - - def _check_input(self, source, destination, feature_dict): - """Validation check for inputs of init method""" - if not isinstance(source, str): - raise ValueError("Parameter source must be string.") - check_filename(source, "source") - - if not isinstance(destination, str): - raise ValueError("Parameter destination must be string.") - check_filename(destination, "destination") - - if feature_dict is None or not isinstance(feature_dict, dict): - raise ValueError("Parameter feature_dict is None or not dict.") - - for _, val in feature_dict.items(): - if not isinstance(val, self.tf.io.FixedLenFeature): - raise ValueError("Parameter feature_dict: {} only support FixedLenFeature.".format(feature_dict)) + return mindrecord_schema def _parse_record(self, example): """Returns features for a single example""" @@ -240,6 +243,24 @@ class TFRecordToMR: except self.tf.errors.InvalidArgumentError: raise ValueError("TFRecord feature_dict parameter error.") + def _get_data_from_tfrecord_sample(self, iterator): + """convert tfrecord sample to mindrecord sample""" + ms_dict = {} + sample = iterator.get_next() + for key, val in sample.items(): + cast_key = _cast_name(key) + if cast_key in self.scalar_set: + self._get_data_when_scalar_field(ms_dict, cast_key, key, val) + else: + if not isinstance(val.numpy(), np.ndarray) and not isinstance(val.numpy(), list): + raise ValueError("The response key: {}, value: {} from TFRecord should be a ndarray or list." + .format(key, val)) + # list set + ms_dict[cast_key] = \ + np.asarray(val, _cast_string_type_to_np_type(self.mindrecord_schema[cast_key]["type"])) + return ms_dict + + def tfrecord_iterator(self): """ Yield a dictionary whose keys are fields in schema. @@ -252,20 +273,7 @@ class TFRecordToMR: iterator = dataset.__iter__() while True: try: - ms_dict = {} - sample = iterator.get_next() - for key, val in sample.items(): - cast_key = _cast_name(key) - if cast_key in self.scalar_set: - self._get_data_when_scalar_field(ms_dict, cast_key, key, val) - else: - if not isinstance(val.numpy(), np.ndarray) and not isinstance(val.numpy(), list): - raise ValueError("The response key: {}, value: {} from TFRecord should be a ndarray or " - "list.".format(key, val)) - # list set - ms_dict[cast_key] = \ - np.asarray(val, _cast_string_type_to_np_type(self.mindrecord_schema[cast_key]["type"])) - yield ms_dict + yield self._get_data_from_tfrecord_sample(iterator) except self.tf.errors.OutOfRangeError: break except self.tf.errors.InvalidArgumentError: