# Copyright (c) MONAI Consortium # Licensed 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. import copy import datetime import functools import importlib import json import operator import os import queue import ssl import subprocess import sys import tempfile import time import traceback import unittest import warnings from contextlib import contextmanager from functools import partial, reduce from subprocess import PIPE, Popen from typing import Callable, Optional, Tuple, Union from urllib.error import ContentTooShortError, HTTPError import numpy as np import torch import torch.distributed as dist from monai.apps.utils import download_url from monai.config import NdarrayTensor from monai.config.deviceconfig import USE_COMPILED from monai.config.type_definitions import NdarrayOrTensor from monai.data import create_test_image_2d, create_test_image_3d from monai.data.meta_tensor import MetaTensor, get_track_meta from monai.networks import convert_to_torchscript from monai.utils import optional_import from monai.utils.module import pytorch_after, version_leq from monai.utils.type_conversion import convert_data_type nib, _ = optional_import("nibabel") quick_test_var = "QUICKTEST" _tf32_enabled = None _test_data_config: dict = {} def testing_data_config(*keys): """get _test_data_config[keys0][keys1]...[keysN]""" if not _test_data_config: with open(os.path.join(os.path.dirname(__file__), "testing_data", "data_config.json")) as c: _config = json.load(c) for k, v in _config.items(): _test_data_config[k] = v return reduce(operator.getitem, keys, _test_data_config) def clone(data: NdarrayTensor) -> NdarrayTensor: """ Clone data independent of type. Args: data (NdarrayTensor): This can be a Pytorch Tensor or numpy array. Returns: Any: Cloned data object """ return copy.deepcopy(data) def assert_allclose( actual: NdarrayOrTensor, desired: NdarrayOrTensor, type_test: Union[bool, str] = True, device_test: bool = False, *args, **kwargs, ): """ Assert that types and all values of two data objects are close. Args: actual: Pytorch Tensor or numpy array for comparison. desired: Pytorch Tensor or numpy array to compare against. type_test: whether to test that `actual` and `desired` are both numpy arrays or torch tensors. if type_test == "tensor", it checks whether the `actual` is a torch.tensor or metatensor according to `get_track_meta`. device_test: whether to test the device property. args: extra arguments to pass on to `np.testing.assert_allclose`. kwargs: extra arguments to pass on to `np.testing.assert_allclose`. """ if isinstance(type_test, str) and type_test == "tensor": if get_track_meta(): np.testing.assert_equal(isinstance(actual, MetaTensor), True, "must be a MetaTensor") else: np.testing.assert_equal( isinstance(actual, torch.Tensor) and not isinstance(actual, MetaTensor), True, "must be a torch.Tensor" ) elif type_test: # check both actual and desired are of the same type np.testing.assert_equal(isinstance(actual, np.ndarray), isinstance(desired, np.ndarray), "numpy type") np.testing.assert_equal(isinstance(actual, torch.Tensor), isinstance(desired, torch.Tensor), "torch type") if isinstance(desired, torch.Tensor) or isinstance(actual, torch.Tensor): if device_test: np.testing.assert_equal(str(actual.device), str(desired.device), "torch device check") # type: ignore actual = actual.detach().cpu().numpy() if isinstance(actual, torch.Tensor) else actual desired = desired.detach().cpu().numpy() if isinstance(desired, torch.Tensor) else desired np.testing.assert_allclose(actual, desired, *args, **kwargs) @contextmanager def skip_if_downloading_fails(): try: yield except (ContentTooShortError, HTTPError, ConnectionError) as e: raise unittest.SkipTest(f"error while downloading: {e}") from e except ssl.SSLError as ssl_e: if "decryption failed" in str(ssl_e): raise unittest.SkipTest(f"SSL error while downloading: {ssl_e}") from ssl_e except RuntimeError as rt_e: if "unexpected EOF" in str(rt_e): raise unittest.SkipTest(f"error while downloading: {rt_e}") from rt_e # incomplete download if "network issue" in str(rt_e): raise unittest.SkipTest(f"error while downloading: {rt_e}") from rt_e if "gdown dependency" in str(rt_e): # no gdown installed raise unittest.SkipTest(f"error while downloading: {rt_e}") from rt_e if "md5 check" in str(rt_e): raise unittest.SkipTest(f"error while downloading: {rt_e}") from rt_e if "account limit" in str(rt_e): # HTTP Error 503: Egress is over the account limit raise unittest.SkipTest(f"error while downloading: {rt_e}") from rt_e raise rt_e def test_pretrained_networks(network, input_param, device): with skip_if_downloading_fails(): return network(**input_param).to(device) def test_is_quick(): return os.environ.get(quick_test_var, "").lower() == "true" def is_tf32_env(): """ The environment variable NVIDIA_TF32_OVERRIDE=0 will override any defaults or programmatic configuration of NVIDIA libraries, and consequently, cuBLAS will not accelerate FP32 computations with TF32 tensor cores. """ global _tf32_enabled if _tf32_enabled is None: _tf32_enabled = False if ( torch.cuda.is_available() and not version_leq(f"{torch.version.cuda}", "10.100") and os.environ.get("NVIDIA_TF32_OVERRIDE", "1") != "0" and torch.cuda.device_count() > 0 # at least 11.0 ): try: # with TF32 enabled, the speed is ~8x faster, but the precision has ~2 digits less in the result g_gpu = torch.Generator(device="cuda") g_gpu.manual_seed(2147483647) a_full = torch.randn(1024, 1024, dtype=torch.double, device="cuda", generator=g_gpu) b_full = torch.randn(1024, 1024, dtype=torch.double, device="cuda", generator=g_gpu) _tf32_enabled = (a_full.float() @ b_full.float() - a_full @ b_full).abs().max().item() > 0.001 # 0.1713 except BaseException: pass print(f"tf32 enabled: {_tf32_enabled}") return _tf32_enabled def skip_if_quick(obj): """ Skip the unit tests if environment variable `quick_test_var=true`. For example, the user can skip the relevant tests by setting ``export QUICKTEST=true``. """ is_quick = test_is_quick() return unittest.skipIf(is_quick, "Skipping slow tests")(obj) class SkipIfNoModule: """Decorator to be used if test should be skipped when optional module is not present.""" def __init__(self, module_name): self.module_name = module_name self.module_missing = not optional_import(self.module_name)[1] def __call__(self, obj): return unittest.skipIf(self.module_missing, f"optional module not present: {self.module_name}")(obj) class SkipIfModule: """Decorator to be used if test should be skipped when optional module is present.""" def __init__(self, module_name): self.module_name = module_name self.module_avail = optional_import(self.module_name)[1] def __call__(self, obj): return unittest.skipIf(self.module_avail, f"Skipping because optional module present: {self.module_name}")(obj) def skip_if_no_cpp_extension(obj): """ Skip the unit tests if the cpp extension is not available. """ return unittest.skipUnless(USE_COMPILED, "Skipping cpp extension tests")(obj) def skip_if_no_cuda(obj): """ Skip the unit tests if torch.cuda.is_available is False. """ return unittest.skipUnless(torch.cuda.is_available(), "Skipping CUDA-based tests")(obj) def skip_if_windows(obj): """ Skip the unit tests if platform is win32. """ return unittest.skipIf(sys.platform == "win32", "Skipping tests on Windows")(obj) def skip_if_darwin(obj): """ Skip the unit tests if platform is macOS (Darwin). """ return unittest.skipIf(sys.platform == "darwin", "Skipping tests on macOS/Darwin")(obj) class SkipIfBeforePyTorchVersion: """Decorator to be used if test should be skipped with PyTorch versions older than that given.""" def __init__(self, pytorch_version_tuple): self.min_version = pytorch_version_tuple self.version_too_old = not pytorch_after(*pytorch_version_tuple) def __call__(self, obj): return unittest.skipIf( self.version_too_old, f"Skipping tests that fail on PyTorch versions before: {self.min_version}" )(obj) class SkipIfAtLeastPyTorchVersion: """Decorator to be used if test should be skipped with PyTorch versions newer than or equal to that given.""" def __init__(self, pytorch_version_tuple): self.max_version = pytorch_version_tuple self.version_too_new = pytorch_after(*pytorch_version_tuple) def __call__(self, obj): return unittest.skipIf( self.version_too_new, f"Skipping tests that fail on PyTorch versions at least: {self.max_version}" )(obj) def is_main_test_process(): ps = torch.multiprocessing.current_process() if not ps or not hasattr(ps, "name"): return False return ps.name.startswith("Main") def has_cupy(): """ Returns True if the user has installed a version of cupy. """ cp, has_cp = optional_import("cupy") if not is_main_test_process(): return has_cp # skip the check if we are running in subprocess if not has_cp: return False try: # test cupy installation with a basic example x = cp.arange(6, dtype="f").reshape(2, 3) y = cp.arange(3, dtype="f") kernel = cp.ElementwiseKernel( "float32 x, float32 y", "float32 z", """ if (x - 2 > y) { z = x * y; } else { z = x + y; } """, "my_kernel" ) flag = kernel(x, y)[0, 0] == 0 del x, y, kernel cp.get_default_memory_pool().free_all_blocks() return flag except Exception: return False HAS_CUPY = has_cupy() def make_nifti_image(array: NdarrayOrTensor, affine=None, dir=None, fname=None, suffix=".nii.gz", verbose=False): """ Create a temporary nifti image on the disk and return the image name. User is responsible for deleting the temporary file when done with it. """ if isinstance(array, torch.Tensor): array, *_ = convert_data_type(array, np.ndarray) if isinstance(affine, torch.Tensor): affine, *_ = convert_data_type(affine, np.ndarray) if affine is None: affine = np.eye(4) test_image = nib.Nifti1Image(array, affine) # if dir not given, create random. Else, make sure it exists. if dir is None: dir = tempfile.mkdtemp() else: os.makedirs(dir, exist_ok=True) # If fname not given, get random one. Else, concat dir, fname and suffix. if fname is None: temp_f, fname = tempfile.mkstemp(suffix=suffix, dir=dir) os.close(temp_f) else: fname = os.path.join(dir, fname + suffix) nib.save(test_image, fname) if verbose: print(f"File written: {fname}.") return fname def make_rand_affine(ndim: int = 3, random_state: Optional[np.random.RandomState] = None): """Create random affine transformation (with values == -1, 0 or 1).""" rs = np.random.random.__self__ if random_state is None else random_state # type: ignore vals = rs.choice([-1, 1], size=ndim) positions = rs.choice(range(ndim), size=ndim, replace=False) af = np.zeros([ndim + 1, ndim + 1]) af[ndim, ndim] = 1 for i, (v, p) in enumerate(zip(vals, positions)): af[i, p] = v return af class DistTestCase(unittest.TestCase): """ testcase without _outcome, so that it's picklable. """ def __getstate__(self): self_dict = self.__dict__.copy() del self_dict["_outcome"] return self_dict def __setstate__(self, data_dict): self.__dict__.update(data_dict) class DistCall: """ Wrap a test case so that it will run in multiple processes on a single machine using `torch.distributed`. It is designed to be used with `tests.utils.DistTestCase`. Usage: decorate a unittest testcase method with a `DistCall` instance:: class MyTests(unittest.TestCase): @DistCall(nnodes=1, nproc_per_node=3, master_addr="localhost") def test_compute(self): ... the `test_compute` method should trigger different worker logic according to `dist.get_rank()`. Multi-node tests require a fixed master_addr:master_port, with node_rank set manually in multiple scripts or from environment variable "NODE_RANK". """ def __init__( self, nnodes: int = 1, nproc_per_node: int = 1, master_addr: str = "localhost", master_port: Optional[int] = None, node_rank: Optional[int] = None, timeout=60, init_method=None, backend: Optional[str] = None, daemon: Optional[bool] = None, method: Optional[str] = "spawn", verbose: bool = False, ): """ Args: nnodes: The number of nodes to use for distributed call. nproc_per_node: The number of processes to call on each node. master_addr: Master node (rank 0)'s address, should be either the IP address or the hostname of node 0. master_port: Master node (rank 0)'s free port. node_rank: The rank of the node, this could be set via environment variable "NODE_RANK". timeout: Timeout for operations executed against the process group. init_method: URL specifying how to initialize the process group. Default is "env://" or "file:///d:/a_temp" (windows) if unspecified. If ``"no_init"``, the `dist.init_process_group` must be called within the code to be tested. backend: The backend to use. Depending on build-time configurations, valid values include ``mpi``, ``gloo``, and ``nccl``. daemon: the process’s daemon flag. When daemon=None, the initial value is inherited from the creating process. method: set the method which should be used to start a child process. method can be 'fork', 'spawn' or 'forkserver'. verbose: whether to print NCCL debug info. """ self.nnodes = int(nnodes) self.nproc_per_node = int(nproc_per_node) if self.nnodes < 1 or self.nproc_per_node < 1: raise ValueError( f"number of nodes and processes per node must be >= 1, got {self.nnodes} and {self.nproc_per_node}" ) self.node_rank = int(os.environ.get("NODE_RANK", "0")) if node_rank is None else int(node_rank) self.master_addr = master_addr self.master_port = np.random.randint(10000, 20000) if master_port is None else master_port if backend is None: self.backend = "nccl" if torch.distributed.is_nccl_available() and torch.cuda.is_available() else "gloo" else: self.backend = backend self.init_method = init_method if self.init_method is None and sys.platform == "win32": self.init_method = "file:///d:/a_temp" self.timeout = datetime.timedelta(0, timeout) self.daemon = daemon self.method = method self.verbose = verbose def run_process(self, func, local_rank, args, kwargs, results): _env = os.environ.copy() # keep the original system env try: os.environ["MASTER_ADDR"] = self.master_addr os.environ["MASTER_PORT"] = str(self.master_port) os.environ["LOCAL_RANK"] = str(local_rank) if self.verbose: os.environ["NCCL_DEBUG"] = "INFO" os.environ["NCCL_DEBUG_SUBSYS"] = "ALL" os.environ["NCCL_BLOCKING_WAIT"] = str(1) os.environ["OMP_NUM_THREADS"] = str(1) os.environ["WORLD_SIZE"] = str(self.nproc_per_node * self.nnodes) os.environ["RANK"] = str(self.nproc_per_node * self.node_rank + local_rank) if torch.cuda.is_available(): torch.cuda.set_device(int(local_rank)) # using device ids from CUDA_VISIBILE_DEVICES if self.init_method != "no_init": dist.init_process_group( backend=self.backend, init_method=self.init_method, timeout=self.timeout, world_size=int(os.environ["WORLD_SIZE"]), rank=int(os.environ["RANK"]), ) func(*args, **kwargs) # the primary node lives longer to # avoid _store_based_barrier, RuntimeError: Broken pipe # as the TCP store daemon is on the rank 0 if int(os.environ["RANK"]) == 0: time.sleep(0.1) results.put(True) except Exception as e: results.put(False) raise e finally: os.environ.clear() os.environ.update(_env) try: dist.destroy_process_group() except RuntimeError as e: warnings.warn(f"While closing process group: {e}.") def __call__(self, obj): if not torch.distributed.is_available(): return unittest.skipIf(True, "Skipping distributed tests because not torch.distributed.is_available()")(obj) if torch.cuda.is_available() and torch.cuda.device_count() < self.nproc_per_node: return unittest.skipIf( True, f"Skipping distributed tests because it requires {self.nnodes} devices " f"but got {torch.cuda.device_count()}", )(obj) _cache_original_func(obj) @functools.wraps(obj) def _wrapper(*args, **kwargs): tmp = torch.multiprocessing.get_context(self.method) processes = [] results = tmp.Queue() func = _call_original_func args = [obj.__name__, obj.__module__] + list(args) for proc_rank in range(self.nproc_per_node): p = tmp.Process( target=self.run_process, args=(func, proc_rank, args, kwargs, results), daemon=self.daemon ) p.start() processes.append(p) for p in processes: p.join() assert results.get(), "Distributed call failed." _del_original_func(obj) return _wrapper class TimedCall: """ Wrap a test case so that it will run in a new process, raises a TimeoutError if the decorated method takes more than `seconds` to finish. It is designed to be used with `tests.utils.DistTestCase`. """ def __init__( self, seconds: float = 60.0, daemon: Optional[bool] = None, method: Optional[str] = "spawn", force_quit: bool = True, skip_timing=False, ): """ Args: seconds: timeout seconds. daemon: the process’s daemon flag. When daemon=None, the initial value is inherited from the creating process. method: set the method which should be used to start a child process. method can be 'fork', 'spawn' or 'forkserver'. force_quit: whether to terminate the child process when `seconds` elapsed. skip_timing: whether to skip the timing constraint. this is useful to include some system conditions such as `torch.cuda.is_available()`. """ self.timeout_seconds = seconds self.daemon = daemon self.force_quit = force_quit self.skip_timing = skip_timing self.method = method @staticmethod def run_process(func, args, kwargs, results): try: output = func(*args, **kwargs) results.put(output) except Exception as e: e.traceback = traceback.format_exc() results.put(e) def __call__(self, obj): if self.skip_timing: return obj _cache_original_func(obj) @functools.wraps(obj) def _wrapper(*args, **kwargs): tmp = torch.multiprocessing.get_context(self.method) func = _call_original_func args = [obj.__name__, obj.__module__] + list(args) results = tmp.Queue() p = tmp.Process(target=TimedCall.run_process, args=(func, args, kwargs, results), daemon=self.daemon) p.start() p.join(timeout=self.timeout_seconds) timeout_error = None try: if p.is_alive(): # create an Exception timeout_error = torch.multiprocessing.TimeoutError( f"'{obj.__name__}' in '{obj.__module__}' did not finish in {self.timeout_seconds}s." ) if self.force_quit: p.terminate() else: warnings.warn( f"TimedCall: deadline ({self.timeout_seconds}s) " f"reached but waiting for {obj.__name__} to finish." ) finally: p.join() _del_original_func(obj) res = None try: res = results.get(block=False) except queue.Empty: # no result returned, took too long pass if isinstance(res, Exception): # other errors from obj if hasattr(res, "traceback"): raise RuntimeError(res.traceback) from res raise res if timeout_error: # no force_quit finished raise timeout_error return res return _wrapper _original_funcs = {} def _cache_original_func(obj) -> None: """cache the original function by name, so that the decorator doesn't shadow it.""" _original_funcs[obj.__name__] = obj def _del_original_func(obj): """pop the original function from cache.""" _original_funcs.pop(obj.__name__, None) if torch.cuda.is_available(): # clean up the cached function torch.cuda.synchronize() torch.cuda.empty_cache() def _call_original_func(name, module, *args, **kwargs): if name not in _original_funcs: _original_module = importlib.import_module(module) # reimport, refresh _original_funcs if not hasattr(_original_module, name): # refresh module doesn't work raise RuntimeError(f"Could not recover the original {name} from {module}: {_original_funcs}.") f = _original_funcs[name] return f(*args, **kwargs) class NumpyImageTestCase2D(unittest.TestCase): im_shape = (128, 64) input_channels = 1 output_channels = 4 num_classes = 3 def setUp(self): im, msk = create_test_image_2d( self.im_shape[0], self.im_shape[1], num_objs=4, rad_max=20, noise_max=0.0, num_seg_classes=self.num_classes ) self.imt = im[None, None] self.seg1 = (msk[None, None] > 0).astype(np.float32) self.segn = msk[None, None] class TorchImageTestCase2D(NumpyImageTestCase2D): def setUp(self): NumpyImageTestCase2D.setUp(self) self.imt = torch.tensor(self.imt) self.seg1 = torch.tensor(self.seg1) self.segn = torch.tensor(self.segn) class NumpyImageTestCase3D(unittest.TestCase): im_shape = (64, 48, 80) input_channels = 1 output_channels = 4 num_classes = 3 def setUp(self): im, msk = create_test_image_3d( self.im_shape[0], self.im_shape[1], self.im_shape[2], num_objs=4, rad_max=20, noise_max=0.0, num_seg_classes=self.num_classes, ) self.imt = im[None, None] self.seg1 = (msk[None, None] > 0).astype(np.float32) self.segn = msk[None, None] class TorchImageTestCase3D(NumpyImageTestCase3D): def setUp(self): NumpyImageTestCase3D.setUp(self) self.imt = torch.tensor(self.imt) self.seg1 = torch.tensor(self.seg1) self.segn = torch.tensor(self.segn) def test_script_save(net, *inputs, device=None, rtol=1e-4, atol=0.0): """ Test the ability to save `net` as a Torchscript object, reload it, and apply inference. The value `inputs` is forward-passed through the original and loaded copy of the network and their results returned. The forward pass for both is done without gradient accumulation. The test will be performed with CUDA if available, else CPU. """ # TODO: would be nice to use GPU if available, but it currently causes CI failures. device = "cpu" with tempfile.TemporaryDirectory() as tempdir: convert_to_torchscript( model=net, filename_or_obj=os.path.join(tempdir, "model.ts"), verify=True, inputs=inputs, device=device, rtol=rtol, atol=atol, ) def download_url_or_skip_test(*args, **kwargs): """``download_url`` and skip the tests if any downloading error occurs.""" with skip_if_downloading_fails(): download_url(*args, **kwargs) def query_memory(n=2): """ Find best n idle devices and return a string of device ids using the `nvidia-smi` command. """ bash_string = "nvidia-smi --query-gpu=power.draw,temperature.gpu,memory.used --format=csv,noheader,nounits" try: p1 = Popen(bash_string.split(), stdout=PIPE) output, error = p1.communicate() free_memory = [x.split(",") for x in output.decode("utf-8").split("\n")[:-1]] free_memory = np.asarray(free_memory, dtype=float).T free_memory[1] += free_memory[0] # combine 0/1 column measures ids = np.lexsort(free_memory)[:n] except (TypeError, IndexError, OSError): ids = range(n) if isinstance(n, int) else [] return ",".join(f"{int(x)}" for x in ids) def test_local_inversion(invertible_xform, to_invert, im, dict_key=None): """test that invertible_xform can bring to_invert back to im""" im_item = im if dict_key is None else im[dict_key] if not isinstance(im_item, MetaTensor): return im_ref = copy.deepcopy(im) im_inv = invertible_xform.inverse(to_invert) if dict_key: im_inv = im_inv[dict_key] im_ref = im_ref[dict_key] np.testing.assert_array_equal(im_inv.applied_operations, []) assert_allclose(im_inv.shape, im_ref.shape) assert_allclose(im_inv.affine, im_ref.affine, atol=1e-3, rtol=1e-3) def command_line_tests(cmd, copy_env=True): test_env = os.environ.copy() if copy_env else os.environ print(f"CUDA_VISIBLE_DEVICES in {__file__}", test_env.get("CUDA_VISIBLE_DEVICES")) try: normal_out = subprocess.run(cmd, env=test_env, check=True, capture_output=True) print(repr(normal_out).replace("\\n", "\n").replace("\\t", "\t")) except subprocess.CalledProcessError as e: output = repr(e.stdout).replace("\\n", "\n").replace("\\t", "\t") errors = repr(e.stderr).replace("\\n", "\n").replace("\\t", "\t") raise RuntimeError(f"subprocess call error {e.returncode}: {errors}, {output}") from e TEST_TORCH_TENSORS: Tuple = (torch.as_tensor,) if torch.cuda.is_available(): gpu_tensor: Callable = partial(torch.as_tensor, device="cuda") TEST_TORCH_TENSORS = TEST_TORCH_TENSORS + (gpu_tensor,) DEFAULT_TEST_AFFINE = torch.tensor( [[2.0, 0.0, 0.0, 0.0], [0.0, 2.0, 0.0, 0.0], [0.0, 0.0, 2.0, 0.0], [0.0, 0.0, 0.0, 1.0]] ) _metatensor_creator = partial(MetaTensor, meta={"a": "b", "affine": DEFAULT_TEST_AFFINE}) TEST_NDARRAYS_NO_META_TENSOR: Tuple[Callable] = (np.array,) + TEST_TORCH_TENSORS # type: ignore TEST_NDARRAYS: Tuple[Callable] = TEST_NDARRAYS_NO_META_TENSOR + (_metatensor_creator,) # type: ignore TEST_TORCH_AND_META_TENSORS: Tuple[Callable] = TEST_TORCH_TENSORS + (_metatensor_creator,) # type: ignore # alias for branch tests TEST_NDARRAYS_ALL = TEST_NDARRAYS TEST_DEVICES = [[torch.device("cpu")]] if torch.cuda.is_available(): TEST_DEVICES.append([torch.device("cuda")]) if __name__ == "__main__": print("\n", query_memory(), sep="\n") # print to stdout sys.exit(0)