Incremental_min_max_calculation_for_large_dataset.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "mount_file_id": "1VckU2z3Y_bIb54ATEcLn7VpfwUOtmLrH",
      "authorship_tag": "ABX9TyNKGvhKdeMmiQnSCPkWmU/P",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    },
    "gpuClass": "standard",
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/ShaswataJash/LargeDatasetHandling/blob/master/Incremental_min_max_calculation_for_large_dataset.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!uname -a\n",
        "!python --version"
      ],
      "metadata": {
        "id": "ABv1EfkN9JmI",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "6fa7b46d-b358-4085-ca16-b249f57be8f5"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Linux 8591b8ca8e76 5.10.147+ #1 SMP Sat Dec 10 16:00:40 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux\n",
            "Python 3.9.16\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!lscpu"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "uj7zmrc1heKe",
        "outputId": "17201b1f-84eb-4ef3-d502-031f694d8e8e"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Architecture:                    x86_64\n",
            "CPU op-mode(s):                  32-bit, 64-bit\n",
            "Byte Order:                      Little Endian\n",
            "Address sizes:                   46 bits physical, 48 bits virtual\n",
            "CPU(s):                          2\n",
            "On-line CPU(s) list:             0,1\n",
            "Thread(s) per core:              2\n",
            "Core(s) per socket:              1\n",
            "Socket(s):                       1\n",
            "NUMA node(s):                    1\n",
            "Vendor ID:                       GenuineIntel\n",
            "CPU family:                      6\n",
            "Model:                           85\n",
            "Model name:                      Intel(R) Xeon(R) CPU @ 2.00GHz\n",
            "Stepping:                        3\n",
            "CPU MHz:                         2000.180\n",
            "BogoMIPS:                        4000.36\n",
            "Hypervisor vendor:               KVM\n",
            "Virtualization type:             full\n",
            "L1d cache:                       32 KiB\n",
            "L1i cache:                       32 KiB\n",
            "L2 cache:                        1 MiB\n",
            "L3 cache:                        38.5 MiB\n",
            "NUMA node0 CPU(s):               0,1\n",
            "Vulnerability Itlb multihit:     Not affected\n",
            "Vulnerability L1tf:              Mitigation; PTE Inversion\n",
            "Vulnerability Mds:               Vulnerable; SMT Host state unknown\n",
            "Vulnerability Meltdown:          Vulnerable\n",
            "Vulnerability Mmio stale data:   Vulnerable\n",
            "Vulnerability Retbleed:          Vulnerable\n",
            "Vulnerability Spec store bypass: Vulnerable\n",
            "Vulnerability Spectre v1:        Vulnerable: __user pointer sanitization and use\n",
            "                                 rcopy barriers only; no swapgs barriers\n",
            "Vulnerability Spectre v2:        Vulnerable, IBPB: disabled, STIBP: disabled, PB\n",
            "                                 RSB-eIBRS: Not affected\n",
            "Vulnerability Srbds:             Not affected\n",
            "Vulnerability Tsx async abort:   Vulnerable\n",
            "Flags:                           fpu vme de pse tsc msr pae mce cx8 apic sep mtr\n",
            "                                 r pge mca cmov pat pse36 clflush mmx fxsr sse s\n",
            "                                 se2 ss ht syscall nx pdpe1gb rdtscp lm constant\n",
            "                                 _tsc rep_good nopl xtopology nonstop_tsc cpuid \n",
            "                                 tsc_known_freq pni pclmulqdq ssse3 fma cx16 pci\n",
            "                                 d sse4_1 sse4_2 x2apic movbe popcnt aes xsave a\n",
            "                                 vx f16c rdrand hypervisor lahf_lm abm 3dnowpref\n",
            "                                 etch invpcid_single ssbd ibrs ibpb stibp fsgsba\n",
            "                                 se tsc_adjust bmi1 hle avx2 smep bmi2 erms invp\n",
            "                                 cid rtm mpx avx512f avx512dq rdseed adx smap cl\n",
            "                                 flushopt clwb avx512cd avx512bw avx512vl xsaveo\n",
            "                                 pt xsavec xgetbv1 xsaves arat md_clear arch_cap\n",
            "                                 abilities\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import torch\n",
        "print(torch.__version__)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "UQ2e2iY8Sb7I",
        "outputId": "fb0f9d85-6a44-4016-b98d-f331946bc5e9"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "2.0.0+cu118\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!df -h /dev/shm"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "FNOT1CsxHpwS",
        "outputId": "6bf249ca-8d78-4836-f1d7-655a3a478a9f"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Filesystem      Size  Used Avail Use% Mounted on\n",
            "shm             5.7G     0  5.7G   0% /dev/shm\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#https://stackoverflow.com/questions/7878707/how-to-unmount-a-busy-device\n",
        "#for python multipprocessor, data across child process and main process are being shared through shared memory\n",
        "#for pytorch Dataloader, shared memory requirement can be quite high\n",
        "!sudo umount -l /dev/shm/ && sudo mount -t tmpfs -o rw,nosuid,nodev,noexec,relatime,size=9G shm /dev/shm"
      ],
      "metadata": {
        "id": "ik3UCDFtQnAs"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#refer: https://numpy.org/doc/stable/reference/global_state.html#madvise-hugepage-on-linux\n",
        "!cat /sys/kernel/mm/transparent_hugepage/enabled\n",
        "!cat /sys/kernel/mm/transparent_hugepage/defrag\n",
        "!cat /sys/kernel/mm/transparent_hugepage/use_zero_page\n",
        "!cat /sys/kernel/mm/transparent_hugepage/hpage_pmd_size"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "R4r4JFDEV7SM",
        "outputId": "2a819bec-bc82-4f02-a05a-2335a28254b5"
      },
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "always [madvise] never\n",
            "always defer defer+madvise [madvise] never\n",
            "1\n",
            "2097152\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%env"
      ],
      "metadata": {
        "id": "dbm8LTIEZS5S",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f43df4ef-e2bb-4d72-db51-4c32791eb586"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "{'SHELL': '/bin/bash',\n",
              " 'NV_LIBCUBLAS_VERSION': '11.11.3.6-1',\n",
              " 'NVIDIA_VISIBLE_DEVICES': 'all',\n",
              " 'COLAB_JUPYTER_TRANSPORT': 'ipc',\n",
              " 'NV_NVML_DEV_VERSION': '11.8.86-1',\n",
              " 'NV_CUDNN_PACKAGE_NAME': 'libcudnn8',\n",
              " 'CGROUP_MEMORY_EVENTS': '/sys/fs/cgroup/memory.events /var/colab/cgroup/jupyter-children/memory.events',\n",
              " 'NV_LIBNCCL_DEV_PACKAGE': 'libnccl-dev=2.16.2-1+cuda11.8',\n",
              " 'NV_LIBNCCL_DEV_PACKAGE_VERSION': '2.16.2-1',\n",
              " 'VM_GCE_METADATA_HOST': '169.254.169.253',\n",
              " 'HOSTNAME': '25fd51ec9744',\n",
              " 'TBE_RUNTIME_ADDR': '172.28.0.1:8011',\n",
              " 'GCE_METADATA_TIMEOUT': '3',\n",
              " 'NVIDIA_REQUIRE_CUDA': 'cuda>=11.8 brand=tesla,driver>=450,driver<451 brand=tesla,driver>=470,driver<471 brand=unknown,driver>=470,driver<471 brand=nvidia,driver>=470,driver<471 brand=nvidiartx,driver>=470,driver<471 brand=geforce,driver>=470,driver<471 brand=geforcertx,driver>=470,driver<471 brand=quadro,driver>=470,driver<471 brand=quadrortx,driver>=470,driver<471 brand=titan,driver>=470,driver<471 brand=titanrtx,driver>=470,driver<471 brand=tesla,driver>=510,driver<511 brand=unknown,driver>=510,driver<511 brand=nvidia,driver>=510,driver<511 brand=nvidiartx,driver>=510,driver<511 brand=geforce,driver>=510,driver<511 brand=geforcertx,driver>=510,driver<511 brand=quadro,driver>=510,driver<511 brand=quadrortx,driver>=510,driver<511 brand=titan,driver>=510,driver<511 brand=titanrtx,driver>=510,driver<511 brand=tesla,driver>=515,driver<516 brand=unknown,driver>=515,driver<516 brand=nvidia,driver>=515,driver<516 brand=nvidiartx,driver>=515,driver<516 brand=geforce,driver>=515,driver<516 brand=geforcertx,driver>=515,driver<516 brand=quadro,driver>=515,driver<516 brand=quadrortx,driver>=515,driver<516 brand=titan,driver>=515,driver<516 brand=titanrtx,driver>=515,driver<516',\n",
              " 'NV_LIBCUBLAS_DEV_PACKAGE': 'libcublas-dev-11-8=11.11.3.6-1',\n",
              " 'NV_NVTX_VERSION': '11.8.86-1',\n",
              " 'COLAB_JUPYTER_IP': '172.28.0.12',\n",
              " 'NV_CUDA_CUDART_DEV_VERSION': '11.8.89-1',\n",
              " 'NV_LIBCUSPARSE_VERSION': '11.7.5.86-1',\n",
              " 'COLAB_LANGUAGE_SERVER_PROXY_ROOT_URL': 'http://172.28.0.1:8013/',\n",
              " 'NV_LIBNPP_VERSION': '11.8.0.86-1',\n",
              " 'NCCL_VERSION': '2.16.2-1',\n",
              " 'KMP_LISTEN_PORT': '6000',\n",
              " 'TF_FORCE_GPU_ALLOW_GROWTH': 'true',\n",
              " 'ENV': '/root/.bashrc',\n",
              " 'PWD': '/',\n",
              " 'TBE_EPHEM_CREDS_ADDR': '172.28.0.1:8009',\n",
              " 'COLAB_LANGUAGE_SERVER_PROXY_REQUEST_TIMEOUT': '30s',\n",
              " 'TBE_CREDS_ADDR': '172.28.0.1:8008',\n",
              " 'NV_CUDNN_PACKAGE': 'libcudnn8=8.7.0.84-1+cuda11.8',\n",
              " 'NVIDIA_DRIVER_CAPABILITIES': 'compute,utility',\n",
              " 'LAST_FORCED_REBUILD': '20230403',\n",
              " 'NV_NVPROF_DEV_PACKAGE': 'cuda-nvprof-11-8=11.8.87-1',\n",
              " 'NV_LIBNPP_PACKAGE': 'libnpp-11-8=11.8.0.86-1',\n",
              " 'NV_LIBNCCL_DEV_PACKAGE_NAME': 'libnccl-dev',\n",
              " 'TCLLIBPATH': '/usr/share/tcltk/tcllib1.20',\n",
              " 'NV_LIBCUBLAS_DEV_VERSION': '11.11.3.6-1',\n",
              " 'NVIDIA_PRODUCT_NAME': 'CUDA',\n",
              " 'COLAB_KERNEL_MANAGER_PROXY_HOST': '172.28.0.12',\n",
              " 'NV_LIBCUBLAS_DEV_PACKAGE_NAME': 'libcublas-dev-11-8',\n",
              " 'NV_CUDA_CUDART_VERSION': '11.8.89-1',\n",
              " 'HOME': '/root',\n",
              " 'LANG': 'en_US.UTF-8',\n",
              " 'CUDA_VERSION': '11.8.0',\n",
              " 'CLOUDSDK_CONFIG': '/content/.config',\n",
              " 'NV_LIBCUBLAS_PACKAGE': 'libcublas-11-8=11.11.3.6-1',\n",
              " 'COLAB_RELEASE_TAG': 'release-colab-20230419-060138-RC00',\n",
              " 'KMP_TARGET_PORT': '9000',\n",
              " 'KMP_EXTRA_ARGS': '--listen_host=172.28.0.12 --target_host=172.28.0.12 --tunnel_background_save_url=https://colab.research.google.com/tun/m/cc48301118ce562b961b3c22d803539adc1e0c19/gpu-t4-s-1yz2m86d6teig --tunnel_background_save_delay=10s --tunnel_periodic_background_save_frequency=30m0s --enable_output_coalescing=true --output_coalescing_required=true',\n",
              " 'NV_LIBNPP_DEV_PACKAGE': 'libnpp-dev-11-8=11.8.0.86-1',\n",
              " 'NV_LIBCUBLAS_PACKAGE_NAME': 'libcublas-11-8',\n",
              " 'COLAB_KERNEL_MANAGER_PROXY_PORT': '6000',\n",
              " 'CLOUDSDK_PYTHON': 'python3',\n",
              " 'NV_LIBNPP_DEV_VERSION': '11.8.0.86-1',\n",
              " 'NO_GCE_CHECK': 'False',\n",
              " 'PYTHONPATH': '/env/python',\n",
              " 'SETUPTOOLS_USE_DISTUTILS': 'stdlib',\n",
              " 'NV_LIBCUSPARSE_DEV_VERSION': '11.7.5.86-1',\n",
              " 'LIBRARY_PATH': '/usr/local/cuda/lib64/stubs',\n",
              " 'NV_CUDNN_VERSION': '8.7.0.84',\n",
              " 'SHLVL': '0',\n",
              " 'NV_CUDA_LIB_VERSION': '11.8.0-1',\n",
              " 'COLAB_LANGUAGE_SERVER_PROXY': '/usr/colab/bin/language_service',\n",
              " 'NVARCH': 'x86_64',\n",
              " 'NV_CUDNN_PACKAGE_DEV': 'libcudnn8-dev=8.7.0.84-1+cuda11.8',\n",
              " 'NV_CUDA_COMPAT_PACKAGE': 'cuda-compat-11-8',\n",
              " 'NV_LIBNCCL_PACKAGE': 'libnccl2=2.16.2-1+cuda11.8',\n",
              " 'LD_LIBRARY_PATH': '/usr/lib64-nvidia',\n",
              " 'COLAB_GPU': '1',\n",
              " 'GCS_READ_CACHE_BLOCK_SIZE_MB': '16',\n",
              " 'NV_NVPROF_VERSION': '11.8.87-1',\n",
              " 'PATH': '/opt/bin:/usr/local/nvidia/bin:/usr/local/cuda/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/tools/node/bin:/tools/google-cloud-sdk/bin',\n",
              " 'NV_LIBNCCL_PACKAGE_NAME': 'libnccl2',\n",
              " 'COLAB_DEBUG_ADAPTER_MUX_PATH': '/usr/local/bin/dap_multiplexer',\n",
              " 'NV_LIBNCCL_PACKAGE_VERSION': '2.16.2-1',\n",
              " 'PYTHONWARNINGS': 'ignore:::pip._internal.cli.base_command',\n",
              " 'DEBIAN_FRONTEND': 'noninteractive',\n",
              " 'COLAB_BACKEND_VERSION': 'next',\n",
              " 'OLDPWD': '/',\n",
              " 'JPY_PARENT_PID': '103',\n",
              " 'TERM': 'xterm-color',\n",
              " 'CLICOLOR': '1',\n",
              " 'PAGER': 'cat',\n",
              " 'GIT_PAGER': 'cat',\n",
              " 'MPLBACKEND': 'module://ipykernel.pylab.backend_inline',\n",
              " 'ENABLE_DIRECTORYPREFETCHER': '1',\n",
              " 'USE_AUTH_EPHEM': '1',\n",
              " 'PYDEVD_USE_FRAME_EVAL': 'NO',\n",
              " 'NUMPY_MADVISE_HUGEPAGE': '1'}"
            ]
          },
          "metadata": {},
          "execution_count": 3
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#https://stackoverflow.com/questions/37890898/how-to-set-env-variable-in-jupyter-notebook\n",
        "%env NUMPY_MADVISE_HUGEPAGE=1"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "OxWhNXIJZec5",
        "outputId": "31419963-8cca-498d-e18e-c87698db26cc"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "env: NUMPY_MADVISE_HUGEPAGE=1\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#Determine total availiable GPU memory"
      ],
      "metadata": {
        "id": "gBnAWDn9AY-H"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "#ref: https://stackoverflow.com/questions/59567226/how-to-programmatically-determine-available-gpu-memory-with-tensorflow\n",
        "import subprocess as sp\n",
        "import os\n",
        "def get_gpu_memory():\n",
        "    command = \"nvidia-smi --query-gpu=memory.free --format=csv\"\n",
        "    try:\n",
        "        memory_free_info = sp.check_output(command.split()).decode('ascii').split('\\n')[:-1][1:]\n",
        "        memory_free_values = [int(x.split()[0]) for i, x in enumerate(memory_free_info)]\n",
        "        return memory_free_values[0] * 1024 * 1024 # memory_free_values[0] is in MB, thus converting into bytes\n",
        "    except Exception as e:\n",
        "        print(e)\n",
        "        return -1"
      ],
      "metadata": {
        "id": "7j4O2tiG7PD_"
      },
      "execution_count": 1,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "#downloading kaggle competitions files"
      ],
      "metadata": {
        "id": "pGV5IETWGxjY"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install kaggle==1.5.12"
      ],
      "metadata": {
        "id": "fmJBzixSfGEN"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NmxlJwRZbYtA"
      },
      "outputs": [],
      "source": [
        "%%python\n",
        "\n",
        "import sys\n",
        "import logging\n",
        "import os\n",
        "import subprocess\n",
        "\n",
        "logging.basicConfig(level=logging.DEBUG, format='%(asctime)s:%(levelname)s:%(message)s')\n",
        "logger = logging.getLogger('my_logger')\n",
        "#handling of kaggle interaction\n",
        "try:\n",
        "    os.environ[\"KAGGLE_CONFIG_DIR\"] = '/home' #kaggle.json file should be uploaded to /home location before executing this cell\n",
        "    kaggle_write_cmd = \"kaggle competitions download -c open-problems-multimodal\"\n",
        "    kaggle_write_call = subprocess.run(kaggle_write_cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)\n",
        "    logger.info(kaggle_write_call.stdout)\n",
        "    if kaggle_write_call.returncode != 0:\n",
        "        logger.error(\"Error in kaggle download, errorcode=%s\", kaggle_write_call.returncode)\n",
        "        sys.stdout.flush()\n",
        "        sys.exit(\"Forceful exit as kaggle download returned error\")\n",
        "except BaseException as err:\n",
        "    logger.error(\"kaggle download related error\", exc_info=True)\n",
        "    sys.stdout.flush()\n",
        "    sys.exit(\"Forceful exit as exception encountered while kaggle download\")"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!mkdir /content/drive/MyDrive/colab_exp_result/kaggle_data\n",
        "!unzip /content/open-problems-multimodal.zip -d /content/drive/MyDrive/colab_exp_result/kaggle_data"
      ],
      "metadata": {
        "id": "TT-gWlUYCG7b"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "We can mount Google drive in colab and can copy the kaggle competitions files there. This will help not to run kaggle download code everytime before start of the notebook - it can save lot of time. Instead, everytime we can directly copy the contents from drive into the local filesystem of the underneath VM hosting the notebook."
      ],
      "metadata": {
        "id": "-tmhnBezqNTP"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!nohup cp /content/drive/MyDrive/colab_exp_result/kaggle_data/* /mnt &"
      ],
      "metadata": {
        "id": "e_jqgtMpmsLS",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "26122276-5490-49a6-e402-c3bc78e5e9a7"
      },
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "nohup: appending output to 'nohup.out'\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!ls -l /mnt"
      ],
      "metadata": {
        "id": "C77_Ud1igRen",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "6dc4364c-3a2b-40de-fea3-c59e25bfadfe"
      },
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "total 28181076\n",
            "-rw------- 1 root root  2418406934 Apr 22 02:06 evaluation_ids.csv\n",
            "-rw------- 1 root root      551250 Apr 22 02:06 max_cite_inputs.txt\n",
            "-rw------- 1 root root     5723550 Apr 22 02:06 max_multi_inputs.txt\n",
            "-rw------- 1 root root      234920 Apr 22 02:06 metadata_cite_day_2_donor_27678.csv\n",
            "-rw------- 1 root root     9770334 Apr 22 02:06 metadata.csv\n",
            "-rw------- 1 root root      551250 Apr 22 02:06 min_cite_inputs.txt\n",
            "-rw------- 1 root root     5723550 Apr 22 02:06 min_multi_inputs.txt\n",
            "-rw------- 1 root root   843563244 Apr 22 02:06 sample_submission.csv\n",
            "-rw------- 1 root root   307964530 Apr 22 02:07 test_cite_inputs_day_2_donor_27678.h5\n",
            "-rw------- 1 root root  1704565845 Apr 22 02:07 test_cite_inputs.h5\n",
            "-rw------- 1 root root  6473530657 Apr 22 02:09 test_multi_inputs.h5\n",
            "-rw------- 1 root root  2498128492 Apr 22 02:09 train_cite_inputs.h5\n",
            "-rw------- 1 root root    38539123 Apr 22 02:09 train_cite_targets.h5\n",
            "-rw------- 1 root root 11334840656 Apr 22 02:13 train_multi_inputs.h5\n",
            "-rw------- 1 root root  3215261538 Apr 22 02:14 train_multi_targets.h5\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#Installation of required software packages"
      ],
      "metadata": {
        "id": "4Pu-jAqrasbR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install h5py==3.8.0"
      ],
      "metadata": {
        "id": "U5wZBh7dyDZc",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "1509452a-57e0-49f3-fbe1-682749984973"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Requirement already satisfied: h5py==3.8.0 in /usr/local/lib/python3.9/dist-packages (3.8.0)\n",
            "Requirement already satisfied: numpy>=1.14.5 in /usr/local/lib/python3.9/dist-packages (from h5py==3.8.0) (1.22.4)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#Ref: https://docs.h5py.org/en/stable/mpi.html\n",
        "#check whether parallel version of h5py is availiable\n",
        "!h5cc -showconfig"
      ],
      "metadata": {
        "id": "TN3pxs5zhvGw"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install hdf5plugin~=2.0"
      ],
      "metadata": {
        "id": "GsjShaVYyIg9",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "044628a6-b3e3-4000-bb06-35bf671eb496"
      },
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting hdf5plugin~=2.0\n",
            "  Downloading hdf5plugin-2.3.2-py2.py3-none-manylinux2014_x86_64.whl (5.4 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m5.4/5.4 MB\u001b[0m \u001b[31m3.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: h5py in /usr/local/lib/python3.9/dist-packages (from hdf5plugin~=2.0) (3.8.0)\n",
            "Requirement already satisfied: numpy>=1.14.5 in /usr/local/lib/python3.9/dist-packages (from h5py->hdf5plugin~=2.0) (1.22.4)\n",
            "Installing collected packages: hdf5plugin\n",
            "Successfully installed hdf5plugin-2.3.2\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#HDF5 handling common code"
      ],
      "metadata": {
        "id": "upQT9S7Ea3tK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import h5py\n",
        "import hdf5plugin #without importing this, decompression will not happen by h5py\n",
        "def get_hdf5_dataset_value_key(hdf5_file, debug = 0):\n",
        "    groups = []\n",
        "    def node_visit(name):\n",
        "        groups.append(name)\n",
        "    \n",
        "    hdf5_file.visit(node_visit)\n",
        "    if debug>0: print(hdf5_file, groups)\n",
        "    \n",
        "    for g in groups:\n",
        "        shape = hdf5_file[g].shape if isinstance(hdf5_file[g], h5py._hl.dataset.Dataset) else None\n",
        "        if debug>0: print(g, type(hdf5_file[g]), shape)\n",
        "        if (not shape is None) and (len(shape) == 2):\n",
        "            return g\n",
        "    \n",
        "    return None\n",
        "\n",
        "def get_hdf5_dataset_with_specific_shape(hdf5_file, size, debug = 0):\n",
        "    groups = []\n",
        "    def node_visit(name):\n",
        "        groups.append(name)\n",
        "    \n",
        "    hdf5_file.visit(node_visit)\n",
        "    if debug>0: print(hdf5_file, groups)\n",
        "    \n",
        "    for g in groups:\n",
        "        shape = hdf5_file[g].shape if isinstance(hdf5_file[g], h5py._hl.dataset.Dataset) else None\n",
        "        if debug>0: print(g, type(hdf5_file[g]), shape)\n",
        "        if (not shape is None) and (len(shape) == 1) and (shape[0] == size):\n",
        "            return g\n",
        "    \n",
        "    return None\n",
        "\n",
        "def get_hdf5_info(hdf5_file):\n",
        "    print('root-group file-object name:', hdf5_file.name)\n",
        "    def print_keys(gr, level):\n",
        "        keys = list(gr.keys())\n",
        "        for k in keys:\n",
        "            \n",
        "            if isinstance(gr[k], h5py._hl.group.Group):\n",
        "                print('->'*level, k, gr[k])\n",
        "                print_keys(gr[k], level + 1)\n",
        "            elif isinstance(gr[k], h5py._hl.dataset.Dataset):\n",
        "                print('->'*level, k, gr[k], 'dtype=', gr[k].dtype , 'size=', gr[k].size, 'nbytes=', gr[k].nbytes, \n",
        "                      'maxshape=', gr[k].maxshape, 'chunks=', gr[k].chunks)\n",
        "\n",
        "    print_keys(hdf5_file, 1)\n",
        "\n"
      ],
      "metadata": {
        "id": "rAqMEPw4yQwL"
      },
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import h5py\n",
        "import hdf5plugin #without importing this, decompression will not happen by h5py\n",
        "print('============= TRAIN MULTI INPUT ====================')\n",
        "train_multi_input_file = h5py.File('/mnt/train_multi_inputs.h5') # HDF5 file\n",
        "get_hdf5_info(train_multi_input_file)\n",
        "train_multi_input_file.close()\n",
        "del train_multi_input_file\n",
        "print('============= TEST MULTI INPUT ====================')\n",
        "test_multi_input_file = h5py.File('/mnt/test_multi_inputs.h5') # HDF5 file\n",
        "get_hdf5_info(test_multi_input_file)\n",
        "test_multi_input_file.close()\n",
        "del test_multi_input_file"
      ],
      "metadata": {
        "id": "8m9Y6UQNp5o9",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "e661a182-dace-401e-d4d9-161616ba944f"
      },
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "============= TRAIN MULTI INPUT ====================\n",
            "root-group file-object name: /\n",
            "-> train_multi_inputs <HDF5 group \"/train_multi_inputs\" (4 members)>\n",
            "->-> axis0 <HDF5 dataset \"axis0\": shape (228942,), type \"|S26\"> dtype= |S26 size= 228942 nbytes= 5952492 maxshape= (228942,) chunks= (2520,)\n",
            "->-> axis1 <HDF5 dataset \"axis1\": shape (105942,), type \"|S12\"> dtype= |S12 size= 105942 nbytes= 1271304 maxshape= (105942,) chunks= (5461,)\n",
            "->-> block0_items <HDF5 dataset \"block0_items\": shape (228942,), type \"|S26\"> dtype= |S26 size= 228942 nbytes= 5952492 maxshape= (228942,) chunks= (2520,)\n",
            "->-> block0_values <HDF5 dataset \"block0_values\": shape (105942, 228942), type \"<f4\"> dtype= float32 size= 24254573364 nbytes= 97018293456 maxshape= (105942, 228942) chunks= (1, 228942)\n",
            "============= TEST MULTI INPUT ====================\n",
            "root-group file-object name: /\n",
            "-> test_multi_inputs <HDF5 group \"/test_multi_inputs\" (4 members)>\n",
            "->-> axis0 <HDF5 dataset \"axis0\": shape (228942,), type \"|S26\"> dtype= |S26 size= 228942 nbytes= 5952492 maxshape= (228942,) chunks= (2520,)\n",
            "->-> axis1 <HDF5 dataset \"axis1\": shape (55935,), type \"|S12\"> dtype= |S12 size= 55935 nbytes= 671220 maxshape= (55935,) chunks= (5461,)\n",
            "->-> block0_items <HDF5 dataset \"block0_items\": shape (228942,), type \"|S26\"> dtype= |S26 size= 228942 nbytes= 5952492 maxshape= (228942,) chunks= (2520,)\n",
            "->-> block0_values <HDF5 dataset \"block0_values\": shape (55935, 228942), type \"<f4\"> dtype= float32 size= 12805870770 nbytes= 51223483080 maxshape= (55935, 228942) chunks= (1, 228942)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import h5py\n",
        "import hdf5plugin #without importing this, decompression will not happen by h5py\n",
        "train_mult_input_file = h5py.File('/mnt/train_multi_inputs.h5') # HDF5 file\n",
        "hdf5_input_key = get_hdf5_dataset_value_key(train_mult_input_file, debug=1)"
      ],
      "metadata": {
        "id": "RsVDw1_dlFFW",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "fea17799-4bf3-4900-90ad-bc8017097cc3"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "<HDF5 file \"train_multi_inputs.h5\" (mode r)> ['train_multi_inputs', 'train_multi_inputs/axis0', 'train_multi_inputs/axis1', 'train_multi_inputs/block0_items', 'train_multi_inputs/block0_values']\n",
            "train_multi_inputs <class 'h5py._hl.group.Group'> None\n",
            "train_multi_inputs/axis0 <class 'h5py._hl.dataset.Dataset'> (228942,)\n",
            "train_multi_inputs/axis1 <class 'h5py._hl.dataset.Dataset'> (105942,)\n",
            "train_multi_inputs/block0_items <class 'h5py._hl.dataset.Dataset'> (228942,)\n",
            "train_multi_inputs/block0_values <class 'h5py._hl.dataset.Dataset'> (105942, 228942)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "hdf5_col_name_key = get_hdf5_dataset_with_specific_shape(train_mult_input_file, 228942, debug=1)\n",
        "cols = train_mult_input_file[hdf5_col_name_key]\n",
        "print(cols.shape)\n",
        "from tqdm import tqdm\n",
        "col_name = []\n",
        "for c_id in tqdm(range(cols.shape[0])):\n",
        "    col_name.append(str(cols[c_id], 'UTF-8'))"
      ],
      "metadata": {
        "id": "5E16JH80v7FX"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "*   https://luis-sena.medium.com/sharing-big-numpy-arrays-across-python-processes-abf0dc2a0ab2 (why ray with shared object store is best sol)\n",
        "*   Ref: https://towardsdatascience.com/histogram-on-function-space-4a710241f026\n",
        "*   Ref: https://stackoverflow.com/questions/71844846/is-there-a-faster-way-to-get-correlation-coefficents (fast corr-coef)\n",
        "\n"
      ],
      "metadata": {
        "id": "4aadQ-lstUfI"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Global min and max determination of the raw-inputs (will be used for min-max normalization of the data)"
      ],
      "metadata": {
        "id": "gbP33e4B293Q"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%writefile rawInputDataset.py\n",
        "\n",
        "#read: https://medium.com/@grvsinghal/speed-up-your-python-code-using-multiprocessing-on-windows-and-jupyter-or-ipython-2714b49d6fac\n",
        "#The above medium article explains why task related class (or the function which is being called from multiprocessor)\n",
        "# has to be defined in separate python class for jupyter notebook\n",
        "\n",
        "from h5py import File\n",
        "from h5py import _hl\n",
        "import hdf5plugin #without importing this, decompression will not happen by h5py\n",
        "from numpy import zeros, s_\n",
        "from torch import cuda, device, from_numpy, transpose\n",
        "from torch import min as torch_min\n",
        "from torch import max as torch_max\n",
        "from subprocess import check_output\n",
        "\n",
        "class RawInputDataset:\n",
        "    \n",
        "    #ref: https://stackoverflow.com/questions/59567226/how-to-programmatically-determine-available-gpu-memory-with-tensorflow\n",
        "    def __get_gpu_memory(self):\n",
        "        command = \"nvidia-smi --query-gpu=memory.free --format=csv\"\n",
        "        try:\n",
        "            memory_free_info = check_output(command.split()).decode('ascii').split('\\n')[:-1][1:]\n",
        "            memory_free_values = [int(x.split()[0]) for i, x in enumerate(memory_free_info)]\n",
        "            return memory_free_values[0] * 1024 * 1024 # memory_free_values[0] is in MB, thus converting into bytes\n",
        "        except Exception as e:\n",
        "            return -1\n",
        "\n",
        "    def __get_hdf5_dataset_value_key(self):\n",
        "        groups = []\n",
        "        def node_visit(name):\n",
        "            groups.append(name)\n",
        "    \n",
        "        self.hdf5_input.visit(node_visit)\n",
        "        if self.debug>0: self.logger.info('%s %s', self.hdf5_input, groups)\n",
        "    \n",
        "        for g in groups:\n",
        "            shape = self.hdf5_input[g].shape if isinstance(self.hdf5_input[g], _hl.dataset.Dataset) else None\n",
        "            if self.debug>0: self.logger.info('group-key:%s, type:%s, shape:%s', g, type(self.hdf5_input[g]), shape)\n",
        "            if (not shape is None) and (len(shape) == 2):\n",
        "                return g\n",
        "    \n",
        "        return None\n",
        "\n",
        "    def __init__(self, hdf5_input_path, batch_size, debug, logger):\n",
        "\n",
        "        self.inited = False\n",
        "        if debug > 0: logger.info('pid=%s hdf5_input_path:%s batch_size:%s debug:%s', os.getpid(), hdf5_input_path, batch_size, debug)\n",
        "\n",
        "        assert batch_size >= 1\n",
        "\n",
        "        self.hdf5_input_path = hdf5_input_path\n",
        "        self.batch_size=batch_size\n",
        "        self.debug = debug\n",
        "        self.logger = logger\n",
        "        \n",
        "        #self.hdf5_input = File(self.hdf5_input_path, 'r', driver='stdio')\n",
        "        self.hdf5_input = File(self.hdf5_input_path, 'r')\n",
        "        hdf5_input_key = self.__get_hdf5_dataset_value_key()\n",
        "        self.hdf5_dataset = self.hdf5_input[hdf5_input_key]\n",
        "        self.len = self.hdf5_dataset.shape[0]\n",
        "\n",
        "        self.cuda_device = device(\"cuda:0\" if cuda.is_available and (self.__get_gpu_memory() > 0) else \"cpu\")\n",
        "        self.input = zeros((batch_size,self.hdf5_dataset.shape[1]), dtype=self.hdf5_dataset.dtype)\n",
        "        self.stat_id_consumed = []\n",
        "        if debug > 0: self.logger.info('hdf5 file: %s hdf5 group: %s hdf5 dataset: %s id=%s', self.hdf5_input, hdf5_input_key, self.hdf5_dataset, id(self.hdf5_input))\n",
        "        if debug > 1: self.logger.debug('torch-device:%s self.input:%s', self.cuda_device, self.input.shape)\n",
        "        self.inited =True\n",
        "\n",
        "    def is_inited(self):\n",
        "        return self.inited\n",
        "    \n",
        "    def __len__(self): return self.len   \n",
        "\n",
        "    def __getitem__(self, row):\n",
        "        try:\n",
        "            assert row < self.len\n",
        "            input = self.hdf5_dataset[row]\n",
        "            if self.debug>0:\n",
        "                self.stat_id_consumed.append(row)\n",
        "            if self.debug > 3: self.logger.debug('type of input=%s shape=%s', type(input) , input.shape)\n",
        "            return from_numpy(input).detach().to(self.cuda_device)\n",
        "        except Exception as e:\n",
        "            self.logger.exception('Exception occurred in __getitem__()')\n",
        "            \n",
        "        return None\n",
        "\n",
        "    def get_batch(self, starting_row):\n",
        "        try:\n",
        "            assert starting_row < self.len\n",
        "            end_row = min(starting_row + self.batch_size, self.len)            \n",
        "            #input = self.hdf5_dataset[starting_row:end_row]\n",
        "            input = self.input\n",
        "            if input.shape[0] != (end_row - starting_row): #will happen for the last batch\n",
        "                input = zeros(((end_row - starting_row),self.hdf5_dataset.shape[1]), dtype=self.hdf5_dataset.dtype)\n",
        "            self.hdf5_dataset.read_direct(input, source_sel=s_[starting_row:end_row,:], dest_sel=None)\n",
        "            if self.debug>0:\n",
        "                self.stat_id_consumed.extend(range(starting_row, end_row))\n",
        "            if self.debug > 3: self.logger.debug('type of input=%s shape=%s', type(input) , input.shape)\n",
        "            return from_numpy(input).to(self.cuda_device) #.detach()\n",
        "        except Exception as e:\n",
        "            self.logger.exception('Exception occurred in get_batch()')\n",
        "            \n",
        "        return None\n",
        "\n",
        "    def find_min_max_on_batch(self, starting_row):\n",
        "        try:\n",
        "            data = self.get_batch(starting_row)\n",
        "            if data is None:\n",
        "                return (None, None)\n",
        "            local_min = torch_min(data, dim=0)[0] #we have to find min for each col (so reduction of dim=0)\n",
        "            local_max = torch_max(data, dim=0)[0] #we have to find max for each col (so reduction of dim=0)  \n",
        "            #local_min = torch_min(transpose(data, 0, 1), dim=1)[0] #we have to find min for each row after transpose (so reduction of dim=1)\n",
        "            #local_max = torch_max(transpose(data, 0, 1), dim=1)[0] #we have to find max for each row after transpose (so reduction of dim=1)  \n",
        "            return (local_min, local_max)\n",
        "        except Exception as e:\n",
        "            self.logger.exception('Exception occurred in find_min_max_on_batch()')\n",
        "        \n",
        "        return (None, None)\n",
        "\n",
        "    def reset_stat(self):\n",
        "        if self.debug <= 0:\n",
        "           return\n",
        "        self.stat_id_consumed.clear()    \n",
        "\n",
        "    def __del__(self):\n",
        "        self.hdf5_input.close()\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ghF5RXLIbtsb",
        "outputId": "f3b2cf14-ff3c-4a9f-d913-255fe5f0be6e"
      },
      "execution_count": 104,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Overwriting rawInputDataset.py\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%%writefile custom_mp.py\n",
        "\n",
        "#read: https://medium.com/@grvsinghal/speed-up-your-python-code-using-multiprocessing-on-windows-and-jupyter-or-ipython-2714b49d6fac\n",
        "#The above medium article explains why Dataset related class (or the function which is being called from multiprocessor)\n",
        "# has to be defined in separate python class for jupyter notebook\n",
        "\n",
        "def my_custom_init(hdf5_input_file_path, batch_size, debug):\n",
        "    from rawInputDataset import RawInputDataset\n",
        "    from multiprocessing import current_process\n",
        "    from threading import current_thread\n",
        "    import logging\n",
        "    from os import getpid\n",
        "\n",
        "    logging.basicConfig(filename=f'my_log-{getpid()}.log', filemode='w', format='%(name)s - %(levelname)s - %(message)s')\n",
        "    \n",
        "    global logger\n",
        "    logger=logging.getLogger(f'child-worker-{getpid()}')\n",
        "    logger.setLevel(logging.DEBUG) \n",
        "\n",
        "    try: \n",
        "       # declare global variable (#https://superfastpython.com/multiprocessing-pool-initializer/#Example_of_Accessing_an_Initialized_Variable_in_a_Worker)\n",
        "       global my_rawInputDataset\n",
        "       # assign the global variable\n",
        "       my_rawInputDataset = RawInputDataset(hdf5_input_file_path, batch_size, debug, logger)\n",
        "       # get the current process\n",
        "       process = current_process()\n",
        "       # get the current thread\n",
        "       thread = current_thread()\n",
        "       # report a message\n",
        "       logger.info('Initializing worker pid=%s process=%s, thread=%s, with=%s', getpid(), process.name, thread.name, my_rawInputDataset)\n",
        "    except Exception as e:\n",
        "        logger.exception('Exception occurred in my_custom_init()')\n",
        "\n",
        "\n",
        "def my_custom_min_max_finding_job(starting_row):\n",
        "    try:\n",
        "        global my_rawInputDataset  \n",
        "        return my_rawInputDataset.find_min_max_on_batch(starting_row)\n",
        "    except Exception as e:\n",
        "        global logger\n",
        "        logger.exception('Exception occurred in my_custom_min_max_finding_job()')\n",
        "    \n",
        "    return (None, None)\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "LAt8WFPnCZ3J",
        "outputId": "ff798a3e-27cd-47f8-92b8-58f25c60f80b"
      },
      "execution_count": 32,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Overwriting custom_mp.py\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#https://stackoverflow.com/questions/2918898/prevent-python-from-caching-the-imported-modules\n",
        "#we have to ensure 'rawInputDataset' module is reloaded everytime (helps us to continously change source code of rawInputDataset.py file)\n",
        "%load_ext autoreload"
      ],
      "metadata": {
        "id": "MSHI-U1lQnCJ"
      },
      "execution_count": 15,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "%autoreload 2"
      ],
      "metadata": {
        "id": "Wy-ZSdr2Qxi5"
      },
      "execution_count": 16,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!python3 -m pip install memray"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "oAbGnW3CAzm5",
        "outputId": "84ab39b7-cad5-4d33-fd9b-7ef360196ebc"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting memray\n",
            "  Downloading memray-1.7.0-cp39-cp39-manylinux_2_12_x86_64.manylinux2010_x86_64.whl (3.2 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m3.2/3.2 MB\u001b[0m \u001b[31m50.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: rich>=11.2.0 in /usr/local/lib/python3.9/dist-packages (from memray) (13.3.3)\n",
            "Requirement already satisfied: jinja2>=2.9 in /usr/local/lib/python3.9/dist-packages (from memray) (3.1.2)\n",
            "Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.9/dist-packages (from jinja2>=2.9->memray) (2.1.2)\n",
            "Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /usr/local/lib/python3.9/dist-packages (from rich>=11.2.0->memray) (2.14.0)\n",
            "Requirement already satisfied: markdown-it-py<3.0.0,>=2.2.0 in /usr/local/lib/python3.9/dist-packages (from rich>=11.2.0->memray) (2.2.0)\n",
            "Requirement already satisfied: mdurl~=0.1 in /usr/local/lib/python3.9/dist-packages (from markdown-it-py<3.0.0,>=2.2.0->rich>=11.2.0->memray) (0.1.2)\n",
            "Installing collected packages: memray\n",
            "Successfully installed memray-1.7.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%load_ext memray"
      ],
      "metadata": {
        "id": "vgv-oWy1A_eb"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!rm -f *.log"
      ],
      "metadata": {
        "id": "35GsMlZRDiAo"
      },
      "execution_count": 106,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#%%memray_flamegraph --trace-python-allocators --follow-fork --native --leaks \n",
        "\n",
        "def find_min_max():\n",
        "    \n",
        "    try:\n",
        "        hdf5_input_file_path = '/mnt/train_multi_inputs.h5'\n",
        "        hdf5_input = h5py_File(hdf5_input_file_path, 'r')\n",
        "        hdf5_input_key = get_hdf5_dataset_value_key(hdf5_input)\n",
        "        data_len = hdf5_input[hdf5_input_key].shape[0]\n",
        "        first_elem = hdf5_input[hdf5_input_key][0]\n",
        "        elem_size_in_bytes = first_elem.size * first_elem.itemsize\n",
        "        print('first_elem.size:', first_elem.size, 'first_elem.itemsize:', first_elem.itemsize, 'elem_size_in_bytes:', elem_size_in_bytes)\n",
        "        hdf5_input.close()\n",
        "        del hdf5_input\n",
        "\n",
        "        optimal_batch_size = floor((1024 * 1024 * 1024) / elem_size_in_bytes) #max 1GB of numpy array \n",
        "        #optimal_batch_size = floor((50 * 1024 * 1024) / elem_size_in_bytes) #max 50MB of numpy array \n",
        "        print('optimal_batch_size:', optimal_batch_size)\n",
        "\n",
        "        while(collect() > 0): pass #clean the memory as much as possible\n",
        "\n",
        "        DEBUG = 0\n",
        "\n",
        "        work_arg = []\n",
        "        for s_row in range(0, data_len, optimal_batch_size):\n",
        "        #for s_row in range(0, 6 * optimal_batch_size, optimal_batch_size):\n",
        "            work_arg.append(s_row)\n",
        "        print(\"total task required:\", len(work_arg))\n",
        "        min_t = None\n",
        "        max_t = None\n",
        "        with  get_context('spawn').Pool(os.cpu_count() * 2, initializer=my_custom_init, initargs=(hdf5_input_file_path,optimal_batch_size,DEBUG )) as p:\n",
        "            #For very long iterables using a large value for chunksize can make the job complete much faster than using the default value of 1 \n",
        "            #(refer: https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool.imap_unordered)\n",
        "            result = p.imap_unordered(my_custom_min_max_finding_job, work_arg, chunksize=1)\n",
        "            \n",
        "            for r in tqdm(result):\n",
        "                if (r[0] is None) or (r[1] is None):\n",
        "                    print(\"None result obtained!!!\", flush=True)\n",
        "                    p.terminate()\n",
        "                    break\n",
        "                if not (min_t is None):\n",
        "                    min_t = minimum(min_t, r[0])\n",
        "                    max_t = maximum(max_t, r[1])\n",
        "                else:\n",
        "                    min_t = r[0]\n",
        "                    max_t = r[1]\n",
        "\n",
        "            # close the process pool\n",
        "            p.close()\n",
        "            # wait for all tasks to complete\n",
        "            p.join()\n",
        "            \n",
        "        print('max.shape:', max_t.shape, 'min.shape:', min_t.shape)\n",
        "        print('max_t', max_t, 'min_t:', min_t) #[10.1655, 11.7670,  9.6392,  ..., 10.8656, 13.2418, 12.2571]\n",
        "        return max_t, min_t\n",
        "    except Exception as e:\n",
        "        print(e)\n",
        "        print_exc()\n",
        "    \n",
        "\n",
        "if __name__ == '__main__':\n",
        "\n",
        "    from math import floor\n",
        "    from traceback import print_exc\n",
        "    from multiprocessing import get_context\n",
        "    from custom_mp import my_custom_init, my_custom_min_max_finding_job\n",
        "    from gc import collect, get_stats\n",
        "    from os import getpid\n",
        "    from h5py import File as h5py_File\n",
        "    from torch import minimum, maximum\n",
        "    from tqdm import tqdm\n",
        "    \n",
        "    print('MAIN pid=', getpid())\n",
        "    find_min_max()\n",
        "    collect()\n",
        "    print(get_stats())"
      ],
      "metadata": {
        "id": "t6Lcq3D7qB__",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "18e9888d-4954-439a-b987-5f21bf2d5c4f"
      },
      "execution_count": 105,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "MAIN pid= 246\n",
            "first_elem.size: 228942 first_elem.itemsize: 4 elem_size_in_bytes: 915768\n",
            "optimal_batch_size: 1172\n",
            "total task required: 91\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "91it [01:48,  1.20s/it]\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "max.shape: torch.Size([228942]) min.shape: torch.Size([228942])\n",
            "max_t tensor([10.1655, 11.7670,  9.6392,  ..., 10.8656, 13.2418, 12.2571],\n",
            "       device='cuda:0') min_t: tensor([0., 0., 0.,  ..., 0., 0., 0.], device='cuda:0')\n",
            "[{'collections': 643, 'collected': 2725, 'uncollectable': 0}, {'collections': 51, 'collected': 646, 'uncollectable': 0}, {'collections': 75, 'collected': 2669, 'uncollectable': 0}]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "np.savetxt('/content/drive/MyDrive/colab_exp_result/kaggle_data/max_multi_inputs.txt', max_m.numpy())\n",
        "np.savetxt('/content/drive/MyDrive/colab_exp_result/kaggle_data/min_multi_inputs.txt', min_m.numpy())"
      ],
      "metadata": {
        "id": "vSZvmRLeFs3G"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "In stead of calculating min max of the input, we can read it everytime from a saved location. This will save time in terms of rerunning the min-max finding algorithm."
      ],
      "metadata": {
        "id": "9IGahACWsL8G"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "max_multi = np.float32(np.loadtxt('/content/drive/MyDrive/colab_exp_result/kaggle_data/max_multi_inputs.txt'))\n",
        "min_multi = np.float32(np.loadtxt('/content/drive/MyDrive/colab_exp_result/kaggle_data/min_multi_inputs.txt'))\n",
        "print(max_multi.shape)\n",
        "print(min_multi.shape)"
      ],
      "metadata": {
        "id": "OoBntWye8_v5"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}