diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index ed4d5fc..a4c0abb 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -1,5 +1,6 @@ # FROM mcr.microsoft.com/devcontainers/cpp:1-ubuntu-24.04 -FROM mcr.microsoft.com/devcontainers/go:1-1.24-bookworm +# FROM mcr.microsoft.com/devcontainers/go +FROM golang:tip-trixie # ARG REINSTALL_CMAKE_VERSION_FROM_SOURCE="none" @@ -11,7 +12,7 @@ FROM mcr.microsoft.com/devcontainers/go:1-1.24-bookworm # [Optional] Uncomment this section to install additional packages. # RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \ # && apt-get -y install --no-install-recommends -RUN apt update && apt -y upgrade && export DEBIAN_FRONTEND=noninteractive && apt -y install openmpi-bin openmpi-doc libopenmpi-dev +RUN apt update && apt -y upgrade && export DEBIAN_FRONTEND=noninteractive && apt -y install openmpi-bin openmpi-doc libopenmpi-dev jq WORKDIR /tmp RUN wget https://mvapich.cse.ohio-state.edu/download/mvapich/osu-micro-benchmarks-7.5.1.tar.gz @@ -20,6 +21,12 @@ WORKDIR /tmp/osu-micro-benchmarks-7.5.1 RUN ./configure CC=/usr/bin/mpicc CXX=/usr/bin/mpicxx RUN make && make install RUN go install golang.org/x/tools/cmd/stringer@latest +RUN go install github.com/go-delve/delve/cmd/dlv@latest + +RUN useradd -m -s /bin/bash vscode +ENV GOPATH=/home/vscode/go +ENV PATH=$PATH:/home/vscode/go/bin +USER vscode # RUN update-alternatives --install /usr/bin/clangd clangd /usr/bin/clangd-19 100 diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 210dc1c..0d96519 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -4,22 +4,14 @@ "name": "Go", "build": { "dockerfile": "Dockerfile" + }, + "remoteUser": "vscode", + "customizations": { + "vscode": { + "extensions": [ + "golang.go", + "anthropic.claude-code" + ] + } } - // Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile - // "image": "mcr.microsoft.com/devcontainers/go:1-1.24-bookworm" - - // Features to add to the dev container. More info: https://containers.dev/features. - // "features": {}, - - // Use 'forwardPorts' to make a list of ports inside the container available locally. - // "forwardPorts": [], - - // Use 'postCreateCommand' to run commands after the container is created. - // "postCreateCommand": "go version", - - // Configure tool-specific properties. - // "customizations": {}, - - // Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root. - // "remoteUser": "root" } diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 86d25e6..4d6c7bd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,25 +1,29 @@ on: [push, pull_request] name: Test + jobs: test: - strategy: - matrix: - go-version: [1.24.x] - os: [ubuntu-latest] - runs-on: ${{ matrix.os }} + runs-on: ubuntu-latest steps: - - name: Install Go - uses: actions/setup-go@v2 + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install Stable Go for Bootstrap + uses: actions/setup-go@v6 with: - go-version: ${{ matrix.go-version }} + go-version: 'stable' + + - name: Download Go Tip + run: | + go install golang.org/dl/gotip@latest + gotip download + echo "$HOME/sdk/gotip/bin" >> $GITHUB_PATH + - name: Install OpenMPI - if: matrix.os == 'ubuntu-latest' - run: sudo apt update && sudo apt install -y --no-install-recommends libopenmpi-dev openmpi-common openmpi-bin - - name: Checkout code - uses: actions/checkout@v2 - - name: Install Stringer - run: go install golang.org/x/tools/cmd/stringer@latest - - name: Build and Install - run: make install + run: sudo apt update && sudo apt install -y --no-install-recommends openmpi-bin libopenmpi-dev + + - name: Build + run: go build ./... + - name: Test - run: make test + run: mpirun -n 4 --oversubscribe go test . diff --git a/.gitignore b/.gitignore index 4781087..89c6dff 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,3 @@ xautogencgoflags.go oldtests/ -cmd/latency +cmd/latency/latency diff --git a/LICENSE.md b/LICENSE.md index fbcd816..095fbcb 100644 --- a/LICENSE.md +++ b/LICENSE.md @@ -1,4 +1,4 @@ -Copyright (c) 2021, Seth Bromberger All rights reserved. +Copyright (c) 2026, Seth Bromberger All rights reserved. Please note the following limitation with respect to redistribution: @@ -13,3 +13,8 @@ Redistribution and use in source and binary forms without modification, and use 1. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + + +## Note on the No-Modification Redistribution Clause +The prohibition on redistribution of modified source code is intentional and specific. It is not intended to prevent private modification or internal use: you are free to modify this software for your own purposes. Rather, it is intended to prevent automated or pipeline-driven modification of this source code prior to or during redistribution, such as the insertion of telemetry, tracking, or analytics code by a package registry, build system, or other distribution intermediary, as well as the distribution of modified copies containing malicious code. Any copy of this software that is redistributed must be identical to what was published by the copyright holder. + diff --git a/MIGRATION.md b/MIGRATION.md new file mode 100644 index 0000000..6ea562f --- /dev/null +++ b/MIGRATION.md @@ -0,0 +1,210 @@ +# Migration Guide + +## Migrating from v0.2 to v0.3 + +### Initialization and teardown + +The package-level `Start`, `Stop`, `WorldRank`, `WorldSize`, and `WorldTime` +functions have been replaced by a session object. + +```go +// v0.2 +mpi.Start(false) +defer mpi.Stop() +rank := mpi.WorldRank() +size := mpi.WorldSize() +``` + +```go +// v0.3 +m, err := mpi.Start() +if err != nil { + log.Fatal(err) +} +defer m.Stop() +rank := m.WorldRank() +size := m.WorldSize() +``` + +For threaded initialization: + +```go +// v0.2 +mpi.Start(true) + +// v0.3 +m, err := mpi.StartThreaded() +``` + +### Initialization check + +```go +// v0.2 +mpi.IsOn() + +// v0.3 +mpi.IsInitialized() +``` + +### Communicators + +`NewCommunicator` is now a method on `*MPI` rather than a package-level function. + +```go +// v0.2 +comm := mpi.NewCommunicator(nil) + +// v0.3 +comm := m.NewCommunicator(nil) +``` + +### Send and receive + +All type-specific send/receive methods (`SendBytes`, `SendInt32s`, `SendFloat64s`, +etc.) have been replaced by generic methods. + +```go +// v0.2 +comm.SendFloat64s(vals, toID, tag) +vals, status := comm.RecvFloat64s(fromID, tag) +comm.RecvPreallocFloat64s(vals, fromID, tag) +``` + +```go +// v0.3 +comm.Send(vals, toID, tag) +vals, status := comm.Recv[float64](fromID, tag) +comm.RecvPrealloc(vals, fromID, tag) +``` + +The same pattern applies to all types: `byte`, `int8`, `int16`, `uint16`, +`int32`, `uint32`, `int64`, `uint64`, `float32`, `float64`, `complex64`, +`complex128`. + +### Single-value send/receive + +```go +// v0.2 +comm.SendFloat64(v, toID, tag) +v, status := comm.RecvFloat64(fromID, tag) +``` + +```go +// v0.3 +comm.SendOne(v, toID, tag) +v, status := comm.RecvOne[float64](fromID, tag) +``` + +### Broadcast + +```go +// v0.2 +comm.BcastFloat64s(vals, root) + +// v0.3 +comm.Bcast(vals, root) +``` + +### Reduce and Allreduce + +```go +// v0.2 +comm.ReduceFloat64s(dest, orig, mpi.OpSum, root) +comm.AllreduceFloat64s(dest, orig, mpi.OpSum, root) + +// v0.3 +comm.Reduce(dest, orig, mpi.OpSum, root) +comm.Allreduce(dest, orig, mpi.OpSum) +``` + +Note that `Allreduce` no longer takes a `root` parameter. The v0.2 parameter +was accepted but never used; v0.3 removes it. + +### Status + +`Status` is now a value type. Methods that previously returned `*Status` now +return `Status`. + +`Count` (formerly `GetCount`) is now generic and no longer takes a `DataType` argument, and has +been renamed along with other `Status` methods. + +| v0.2 | v0.3 | +|------|------| +| `Status.GetSource()` | `Status.Source()` | +| `Status.GetTag()` | `Status.Tag()` | +| `Status.GetError()` | `Status.Error()` | +| `Status.GetCount(mpi.Byte)` | `Status.Count[byte]()` | + +### Mprobe and Mrecv + +The v0.2 API exposed `C.MPI_Message` directly. In v0.3 this is encapsulated +in `MatchedMessage`. + +```go +// v0.2 +status, msg := comm.Mprobe(fromID, tag) +n := status.GetCount(mpi.Byte) +buf := make([]byte, n) +comm.MrecvPreallocBytes(buf, fromID, tag, msg) +``` + +```go +// v0.3 +m := comm.Mprobe(fromID, tag) +buf, status := m.Recv[byte]() +// or, with a preallocated buffer: +status = m.RecvPrealloc(buf) +``` + +`Mrecv` is also available as a single call: + +```go +buf, status := comm.Mrecv[byte](fromID, tag) +``` + +`MatchedMessage` methods follow the same naming convention as `Status`: + +| v0.2 | v0.3 | +|------|------| +| `MatchedMessage.GetSource()` | `MatchedMessage.Source()` | +| `MatchedMessage.GetTag()` | `MatchedMessage.Tag()` | +| `MatchedMessage.GetError()` | `MatchedMessage.Error()` | +| `MatchedMessage.GetCount[T]()` | `MatchedMessage.Count[T]()` | + +### Communicator methods + +| v0.2 | v0.3 | +|------|------| +| `Communicator.GetAttr()` | `Communicator.Attr()` | +| `Communicator.GetMaxTag()` | `Communicator.MaxTag()` | + +### Communicator.MaxTag field removed + +The exported `MaxTag` field on `Communicator` has been unexported. Use +`Communicator.MaxTag()` if you need this value. + +### DataType removed + +The exported `DataType` constants (`mpi.Byte`, `mpi.Int`, `mpi.Float`, +`mpi.Double`, etc.) have been removed. Type information is now conveyed through +Go generics and is not part of the public API. + +### Iprobe + +`Iprobe` now returns a value `Status` rather than a pointer. + +### String convenience methods + +`SendString` and `RecvString` have been removed. Use `Send` and `Recv` with a +`[]byte` conversion: + +```go +// v0.2 +comm.SendString(s, toID, tag) +s, status := comm.RecvString(fromID, tag) + +// v0.3 +comm.Send([]byte(s), toID, tag) +b, status := comm.Recv[byte](fromID, tag) +s := string(b) +``` diff --git a/Makefile b/Makefile deleted file mode 100644 index d28f380..0000000 --- a/Makefile +++ /dev/null @@ -1,63 +0,0 @@ -.DEFAULT_GOAL := build - -unamestr := $(shell uname) -INVALID=true -ifeq ($(unamestr), Linux) -INVALID=false -endif -ifeq ($(unamestr), Darwin) -INVALID=false -endif -ifeq ($(unamestr), FreeBSD) -INVALID=false -endif - - -CFLAGS=`mpicc -showme:compile` -LDFLAGS=`mpicc -showme:link` -FLAGS_FILE="xautogencgoflags.go" - -define XGENFLAGS -// Copyright 2021 Seth Bromberger. All rights reserved.\n\ -// Copyright 2016 The Gosl Authors. All rights reserved.\n\ -// Use of this source code is governed by a BSD-style\n\ -// license that can be found in the LICENSE file.\n\ -// *** NOTE: this file was auto generated by Makefile ***\n\ -// *** and should be ignored ***\n\ -\n\ -// +build !windows\n\ -\n\ -package mpi\n\ -/*\n\ -#cgo CFLAGS: $(CFLAGS) \n\ -#cgo LDFLAGS: $(LDFLAGS)\n\ -*/\n\ -import \"C\"\n -endef - -xgenflags: - @echo "$(XGENFLAGS)" | sed "s/^ //g" > $(FLAGS_FILE) - -valid: - @if [ "$(INVALID)" = "true" ]; then echo "Architecture $(unamestr) is not supported by this library"; exit 127; fi - -gobuild: - { \ - touch *.go ; \ - go generate ; \ - } - -goinstall: gobuild - { \ - go install ; \ - } - -build: valid xgenflags - -install: build goinstall - -test: - mpirun -n 4 --oversubscribe go test . ; - -clean: - rm -f $(FLAGS_FILE) diff --git a/README.md b/README.md index c07b212..501cd4e 100644 --- a/README.md +++ b/README.md @@ -2,51 +2,134 @@ GoMPI: Message Passing Interface for Parallel Computing -The `gompi` package is a lightweight wrapper to the [OpenMPI](https://www.open-mpi.org) C++ library -designed to develop algorithms for parallel computing. +The `gompi` package is a lightweight wrapper to the [OpenMPI](https://www.open-mpi.org) C library designed to develop algorithms for parallel computing. GoMPI is a fork of the [gosl](https://github.com/cpmech/gosl) MPI library with additional methods. -## Installation +## Usage -1) install [OpenMPI](https://www.open-mpi.org) for your system -2) ensure [golang.org/x/tools/cmd/stringer](https://godoc.org/golang.org/x/tools/cmd/stringer) is installed (`go install` if not) -3) run `make install` +```go +package main -(Other `make` options include `test`, `build`, and `clean`.) +import ( + "fmt" + "log" + mpi "github.com/sbromberger/gompi" +) -## Performance +func main() { + m, err := mpi.Start() + if err != nil { + log.Fatal(err) + } + defer m.Stop() + + rank := m.WorldRank() + size := m.WorldSize() + + comm := m.NewCommunicator(nil) + + if rank == 0 { + // Rank 0 sends a slice of float64 to rank 1. + vals := []float64{1.0, 2.0, 3.0} + comm.Send(vals, 1, 0) + fmt.Printf("rank 0 of %d: sent %v\n", size, vals) + } else if rank == 1 { + // Rank 1 receives from rank 0. + vals, _ := comm.Recv[float64](0, 0) + fmt.Printf("rank 1 of %d: received %v\n", size, vals) + } +} +``` + +Run with: + +``` +mpirun -n 2 go run main.go +``` + +## Dependencies +**This package will not work on Windows systems.** + +GoMPI requires the [OpenMPI](https://www.open-mpi.org) libraries, header files, and binaries to be installed on your system. + + +## Testing +Testing requires four MPI ranks and is launched via `mpirun`: -Note: latency benchmarks updated August 2025. +``` +mpirun -n 4 --oversubscribe go test . +``` -OSU bechmarks run using `mpirun -n 2 ./osu_latency -i 1000 -x 200` with datatype = `MPI_Char`. +## Performance +Note: latency benchmarks updated May 2026. + +### Single-Node Benchmarks +OSU MPI Latency Test (v7.5.1) benchmarks run using `mpirun -n 2 ./osu_latency -i 1000 -x 200` with datatype = `MPI_Char`. GoMPI benchmarks run using `mpirun -n 2 go run latency.go`. -| message size (bytes) | GoMPI (µs) | OSU MPI Latency Test v7.5 (µs) | -|---|---|---| -| 1 | 0.16 | 0.11 | -| 2 | 0.18 | 0.11 | -| 4 | 0.16 | 0.11 | -| 8 | 0.17 | 0.11 | -| 16 | 0.17 | 0.11 | -| 32 | 0.17 | 0.11 | -| 64 | 0.17 | 0.12 | -| 128 | 0.19 | 0.13 | -| 256 | 0.19 | 0.16 | -| 512 | 0.28 | 0.26 | -| 1024 | 0.29 | 0.28 | -| 2048 | 0.34 | 0.35 | -| 4096 | 0.76 | 0.77 | -| 8192 | 0.92 | 0.88 | -| 16384 | 1.28 | 1.10 | -| 32768 | 1.96 | 1.45 | -| 65536 | 3.07 | 2.59 | -| 131072 | 4.48 | 4.98 | -| 262144 | 7.71 | 7.64 | -| 524288 | 13.71 | 13.25 | -| 1048576 | 25.99 | 25.25 | -| 2097152 | 56.97 | 49.65 | -| 4194304 | 140.32 | 244.79 | - -Benchmark code may be found in `cmd/latency.go`. +Benchmarks were run on a single node. Small-message overhead reflects CGo call latency and converges to parity as message size increases. + +| message size (bytes) | GoMPI (µs) | OSU MPI (µs) | difference | +|---|---|---|---| +| 1 | 0.13 | 0.10 | 1.3x | +| 2 | 0.13 | 0.10 | 1.3x | +| 4 | 0.14 | 0.10 | 1.4x | +| 8 | 0.13 | 0.10 | 1.3x | +| 16 | 0.13 | 0.10 | 1.3x | +| 32 | 0.14 | 0.10 | 1.4x | +| 64 | 0.15 | 0.11 | 1.4x | +| 128 | 0.17 | 0.11 | 1.5x | +| 256 | 0.17 | 0.14 | 1.2x | +| 512 | 0.25 | 0.20 | 1.2x | +| 1024 | 0.28 | 0.23 | 1.2x | +| 2048 | 0.31 | 0.29 | 1.1x | +| 4096 | 0.66 | 0.68 | 1.0x | +| 8192 | 0.88 | 0.91 | 1.0x | +| 16384 | 1.08 | 1.15 | 0.9x | +| 32768 | 1.54 | 1.59 | 1.0x | +| 65536 | 2.92 | 2.27 | 1.3x | +| 131072 | 4.17 | 4.10 | 1.0x | +| 262144 | 7.45 | 6.83 | 1.1x | +| 524288 | 14.00 | 13.22 | 1.1x | +| 1048576 | 26.63 | 24.71 | 1.1x | +| 2097152 | 52.03 | 50.45 | 1.0x | +| 4194304 | 100.70 | 102.49 | 1.0x | + +Benchmark code may be found in `cmd/latency/latency.go`. + +### Inter-node Benchmarks (OmniPath/PSM2) + +OSU MPI Latency Test (v5.9) benchmarks run using `srun -N 2 -n 2 --ntasks-per-node=1 ./osu_latency -i 1000 -x 200` with datatype = `MPI_Char`. +GoMPI benchmarks run using `srun -N 2 -n 2 --ntasks-per-node=1 go run main.go`. + +Benchmarks were run across two nodes over OmniPath (PSM2 transport). Small-message overhead reflects CGo call latency and converges to parity as message size increases. + +| message size (bytes) | GoMPI (µs) | OSU MPI (µs) | difference | +|---|---|---|---| +| 1 | 1.14 | 1.01 | 1.1x | +| 2 | 1.16 | 1.00 | 1.2x | +| 4 | 1.18 | 0.99 | 1.2x | +| 8 | 1.13 | 0.99 | 1.1x | +| 16 | 1.26 | 1.09 | 1.2x | +| 32 | 1.27 | 1.10 | 1.2x | +| 64 | 1.28 | 1.10 | 1.2x | +| 128 | 1.30 | 1.13 | 1.2x | +| 256 | 1.33 | 1.17 | 1.1x | +| 512 | 1.44 | 1.25 | 1.2x | +| 1024 | 1.49 | 1.37 | 1.1x | +| 2048 | 1.72 | 1.59 | 1.1x | +| 4096 | 2.15 | 2.04 | 1.1x | +| 8192 | 2.53 | 2.39 | 1.1x | +| 16384 | 7.15 | 7.03 | 1.0x | +| 32768 | 12.32 | 6.77 | 1.8x¹ | +| 65536 | 11.65 | 11.57 | 1.0x | +| 131072 | 15.80 | 15.81 | 1.0x | +| 262144 | 23.23 | 23.32 | 1.0x | +| 524288 | 36.53 | 36.22 | 1.0x | +| 1048576 | 64.27 | 64.59 | 1.0x | +| 2097152 | 119.08 | 118.28 | 1.0x | +| 4194304 | 211.46 | 212.02 | 1.0x | + +¹ Anomalous result reflecting fabric instability at this message size; not representative of CGo overhead. diff --git a/cmd/latency.go b/cmd/latency/main.go similarity index 60% rename from cmd/latency.go rename to cmd/latency/main.go index 56f6e68..7069dd4 100644 --- a/cmd/latency.go +++ b/cmd/latency/main.go @@ -2,6 +2,7 @@ package main import ( "fmt" + "log" "strings" mpi "github.com/sbromberger/gompi" @@ -14,34 +15,37 @@ const ( ) func main() { - mpi.Start(false) // line 45 - defer mpi.Stop() - o := mpi.NewCommunicator(nil) - if mpi.WorldSize() != 2 { // line 88 + m, err := mpi.Start() + if err != nil { + log.Fatal(err) + } // line 45 + defer m.Stop() + o := m.NewCommunicator(nil) + if m.WorldSize() != 2 { // line 88 panic("This test requires exactly 2 processors") } myId := o.Rank() // line 49 - for size := 1; size <= maxsize; size *=2 { + for size := 1; size <= maxsize; size *= 2 { var t_total float64 s_buf := []byte(strings.Repeat("a", size)) r_buf := []byte(strings.Repeat("b", size)) o.Barrier() - for iter := range (warmup + iterations) { + for iter := range warmup + iterations { notime := iter < warmup switch myId { case 0: - t_start := mpi.WorldTime() // line 140 - o.SendBytes(s_buf, 1, 1) - o.RecvPreallocBytes(s_buf, 1, 1) - t_end := mpi.WorldTime() + t_start := m.WorldTime() // line 140 + o.Send(s_buf, 1, 1) + o.RecvPrealloc(s_buf, 1, 1) + t_end := m.WorldTime() if !notime { t_total += t_end - t_start } case 1: - o.RecvPreallocBytes(r_buf, 0, 1) - o.SendBytes(r_buf, 0, 1) + o.RecvPrealloc(r_buf, 0, 1) + o.Send(r_buf, 0, 1) } diff --git a/datatype_string.go b/datatype_string.go deleted file mode 100644 index 9d4ce01..0000000 --- a/datatype_string.go +++ /dev/null @@ -1,30 +0,0 @@ -// Code generated by "stringer -type=DataType"; DO NOT EDIT. - -package mpi - -import "strconv" - -func _() { - // An "invalid array index" compiler error signifies that the constant values have changed. - // Re-run the stringer command to generate them again. - var x [1]struct{} - _ = x[Byte-0] - _ = x[Uint-1] - _ = x[Int-2] - _ = x[Ulong-3] - _ = x[Long-4] - _ = x[Float-5] - _ = x[Double-6] - _ = x[Complex-7] -} - -const _DataType_name = "ByteUintIntUlongLongFloatDoubleComplex" - -var _DataType_index = [...]uint8{0, 4, 8, 11, 16, 20, 25, 31, 38} - -func (i DataType) String() string { - if i >= DataType(len(_DataType_index)-1) { - return "DataType(" + strconv.FormatInt(int64(i), 10) + ")" - } - return _DataType_name[_DataType_index[i]:_DataType_index[i+1]] -} diff --git a/flags.go b/flags.go index 41605ca..04d02c5 100644 --- a/flags.go +++ b/flags.go @@ -2,15 +2,11 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -// +build !windows +//go:build !windows package mpi /* -#cgo linux CFLAGS: -I/usr/lib/x86_64-linux-gnu/openmpi/include/openmpi -I/usr/lib/x86_64-linux-gnu/openmpi/include -pthread -#cgo linux LDFLAGS: -pthread -L/usr/lib/x86_64-linux-gnu/openmpi/lib -lmpi - -#cgo darwin CFLAGS: -I/usr/local/Cellar/open-mpi/4.0.1_2/include -#cgo darwin LDFLAGS: -L/usr/local/opt/libevent/lib -L/usr/local/Cellar/open-mpi/4.0.1_2/lib -lmpi +#cgo pkg-config: ompi */ import "C" diff --git a/go.mod b/go.mod index 7ef8877..4887ea3 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/sbromberger/gompi -go 1.24 +go 1.27 diff --git a/mpi.go b/mpi.go index 15c48f9..e143bf3 100644 --- a/mpi.go +++ b/mpi.go @@ -1,4 +1,4 @@ -// Copyright 2019 Seth Bromberger. All Rights Reserved. +// Copyright 2026 Seth Bromberger. All Rights Reserved. // This code was derived from / inspired by Gosl: // Copyright 2016 The Gosl Authors. All rights reserved. @@ -6,12 +6,12 @@ // license that can be found in the LICENSE file. //go:build !windows -// +build !windows -//go:generate stringer -type=DataType -//go:generate stringer -type=Op +// Package mpi provides a Go wrapper around the Message Passing Interface (MPI) +// for distributed parallel computation. It supports point-to-point messaging, +// collective operations (broadcast, reduce, allreduce), and communicator +// management. -// Package mpi wraps the Message Passing Interface for parallel computations package mpi /* @@ -26,91 +26,78 @@ import "C" import ( "fmt" - "log" "unsafe" ) -type DataType uint8 - +// AnySource and AnyTag are wildcard values for use in receive operations. const ( AnySource = C.MPI_ANY_SOURCE AnyTag = C.MPI_ANY_TAG ) +// CommTypeShared is the MPI communicator type for processes sharing memory. const ( - // These constants represent (a subset of) MPI datatypes. - Byte DataType = iota - Uint // This maps to a uint32 in go. - Int // This maps to an int32 in go. - Ulong // This maps to a uint64 in go. - Long // This maps to an int64 in go. - Float // This maps to a float32 in go - Double // This maps to a float64 in go. - Complex // This maps to a complex128 in go. + CommTypeShared = C.MPI_COMM_TYPE_SHARED ) -var dataTypes = [...]C.MPI_Datatype{ - C.MPI_BYTE, - C.MPI_UINT32_T, - C.MPI_INT32_T, - C.MPI_UINT64_T, - C.MPI_INT64_T, - C.MPI_FLOAT, - C.MPI_DOUBLE, - C.MPI_DOUBLE_COMPLEX, +// Status holds the result of a completed MPI operation, including the source, +// tag, error code, and element count of the received message. +type Status struct { + mpiStatus C.MPI_Status } -const ( - CommTypeShared = C.MPI_COMM_TYPE_SHARED -) +// MatchedMessage is an atomically claimed message handle returned by Mprobe. +// It carries the probe status and must be consumed by RecvPrealloc or Recv. +type MatchedMessage struct { + msg C.MPI_Message + status Status +} -type Op uint8 +// Source returns the rank of the process that sent this message. +func (m *MatchedMessage) Source() int { return m.status.Source() } -const ( - OpSum Op = iota - OpMin - OpMax - OpProd - OpLand - OpLor - OpLxor - OpBand - OpBor - OpBxor -) +// Tag returns the tag of this message. +func (m *MatchedMessage) Tag() int { return m.status.Tag() } -var ops = [...]C.MPI_Op{ - C.MPI_SUM, - C.MPI_MIN, - C.MPI_MAX, - C.MPI_PROD, - C.MPI_LAND, - C.MPI_LOR, - C.MPI_LXOR, - C.MPI_BAND, - C.MPI_BOR, - C.MPI_BXOR, -} - -// Returns true if the datatype can be used for the given operation. -// This is needed because boolean/logical operators are invalid for non-ints, -// and complex numbers have no ordering. -func isValidDataTypeForOp(d DataType, o Op) bool { - if o == OpLand || o == OpLor || o == OpLxor || o == OpBand || o == OpBor || o == OpBxor { - return d == Byte || d == Uint || d == Int || d == Ulong || d == Long - } - if o == OpMin || o == OpMax { - return d != Complex - } - return true +// Error returns the error code from the probe that claimed this message. +func (m *MatchedMessage) Error() int { return m.status.Error() } + +// Count returns the number of elements of type T in this message. +func (m *MatchedMessage) Count[T goTypes]() int { return m.status.Count[T]() } + +// RecvPrealloc receives the claimed message into the pre-allocated slice buf. +// This is a pointer receiver because MPI_Mrecv modifies the message handle. +func (m *MatchedMessage) RecvPrealloc[T goTypes](buf []T) Status { + var s Status + C.MPI_Mrecv(unsafe.Pointer(unsafe.SliceData(buf)), C.int(len(buf)), dataTypeOf[T](), &m.msg, &s.mpiStatus) + return s } -// Status wraps an MPI_Status structure. -type Status struct { - mpiStatus C.MPI_Status +// Recv allocates and returns a slice containing the claimed message. +// This is a pointer receiver because MPI_Mrecv modifies the message handle. +func (m *MatchedMessage) Recv[T goTypes]() ([]T, Status) { + buf := make([]T, m.Count[T]()) + return buf, m.RecvPrealloc(buf) } -func (o *Communicator) GetAttr(attribute int) (int, bool, error) { +// Mprobe blocks until a message matching source and tag is available, claims +// it atomically, and returns a *MatchedMessage. This is the thread-safe +// alternative to Probe. +func (o *Communicator) Mprobe(source int, tag int) *MatchedMessage { + m := &MatchedMessage{} + C.MPI_Mprobe(C.int(source), C.int(tag), o.comm, &m.msg, &m.status.mpiStatus) + return m +} + +// Mrecv atomically claims and receives a message from fromID with the given tag. +// It is equivalent to calling Mprobe followed by Recv on the returned MatchedMessage. +func (o *Communicator) Mrecv[T goTypes](fromID int, tag int) ([]T, Status) { + return o.Mprobe(fromID, tag).Recv[T]() +} + +// Attr retrieves a communicator attribute by key. It returns the attribute +// value, a boolean indicating whether the attribute was set, and any error. +func (o *Communicator) Attr(attribute int) (int, bool, error) { var n int var found C.int @@ -121,8 +108,9 @@ func (o *Communicator) GetAttr(attribute int) (int, bool, error) { return int(n), int(found) == 1, nil } -func (o *Communicator) GetMaxTag() (int, error) { - x, found, err := o.GetAttr(C.MPI_TAG_UB) +// MaxTag returns the maximum tag value supported by this communicator. +func (o *Communicator) MaxTag() (int, error) { + x, found, err := o.Attr(C.MPI_TAG_UB) if !found { return -1, fmt.Errorf("no max tag value found") } @@ -132,114 +120,130 @@ func (o *Communicator) GetMaxTag() (int, error) { return x, nil } -// Probe issues an MPI Probe and returns a Status structure. -func (o *Communicator) Probe(source int, tag int) *Status { +// Probe blocks until a message matching source and tag is available, and +// returns its status. The message is not consumed; use a Recv to read it. +func (o *Communicator) Probe(source int, tag int) Status { var s Status C.MPI_Probe(C.int(source), C.int(tag), o.comm, &(s.mpiStatus)) - return &s -} - -func (o *Communicator) Mprobe(source int, tag int) (Status, C.MPI_Message) { - var s Status - var msg C.MPI_Message - C.MPI_Mprobe(C.int(source), C.int(tag), o.comm, &msg, &(s.mpiStatus)) - return s, msg + return s } -// GetCount returns a count of elements of type `t` from a Status object. -func (s *Status) GetCount(t DataType) int { +// Count returns the number of elements of type T in the received message +// described by this Status. +func (s Status) Count[T goTypes]() int { var n C.int - C.MPI_Get_count(&s.mpiStatus, dataTypes[t], &n) + C.MPI_Get_count(&s.mpiStatus, dataTypeOf[T](), &n) return int(n) } -// GetError returns the error code from a Status object. -func (s *Status) GetError() int { +// Error returns the error code associated with this Status. +func (s Status) Error() int { return int(s.mpiStatus.MPI_ERROR) } -// GetSource returns the source (sender) of an MPI message. -func (s *Status) GetSource() int { +// Source returns the rank of the processor that sent the message described +// by this Status. +func (s Status) Source() int { return int(s.mpiStatus.MPI_SOURCE) } -// GetTag returns the tag associated with the MPI channel. -func (s *Status) GetTag() int { +// Tag returns the tag of the message described by this Status. +func (s Status) Tag() int { return int(s.mpiStatus.MPI_TAG) } -// IsOn tells whether MPI is on or not -// NOTE: this returns true even after Stop -func IsOn() bool { - var flag C.int - C.MPI_Initialized(&flag) - return flag != 0 +// IsInitialized reports whether MPI has been initialised and not yet finalised. +func IsInitialized() bool { + var init, fin C.int + C.MPI_Initialized(&init) + C.MPI_Finalized(&fin) + return init != 0 && fin == 0 } -// Start initialises MPI -func Start(threaded bool) { - if threaded { - var x C.int - C.MPI_Init_thread(nil, nil, C.MPI_THREAD_MULTIPLE, &x) - if x != C.MPI_THREAD_MULTIPLE { - log.Fatalf("Requested threading support %d not available (%d).", C.MPI_THREAD_MULTIPLE, x) - } - } else { - C.MPI_Init(nil, nil) +// MPI is a token representing an active MPI session. It is obtained by calling +// Start or StartThreaded and must be used to access world-level operations and +// create communicators. Only one MPI session may exist per process. +type MPI struct{} + +// Start initialises MPI and returns a session token. It returns an error if +// MPI is already initialised. MPI's default error handler +// (MPI_ERRORS_ARE_FATAL) will abort the process on any subsequent MPI failure. +func Start() (*MPI, error) { + if IsInitialized() { + return nil, fmt.Errorf("MPI is already initialized") } + C.MPI_Init(nil, nil) + return &MPI{}, nil } -// Stop finalises MPI -func Stop() { +// StartThreaded initialises MPI with full thread support (MPI_THREAD_MULTIPLE) +// and returns a session token. It returns an error if MPI is already +// initialised or if the requested threading level is not available. +func StartThreaded() (*MPI, error) { + if IsInitialized() { + return nil, fmt.Errorf("MPI is already initialized") + } + var x C.int + C.MPI_Init_thread(nil, nil, C.MPI_THREAD_MULTIPLE, &x) + if x != C.MPI_THREAD_MULTIPLE { + return nil, fmt.Errorf("MPI thread support %d unavailable (got %d).", C.MPI_THREAD_MULTIPLE, x) + } + return &MPI{}, nil +} + +// Stop finalises MPI. No MPI calls may be made after Stop returns. +func (m *MPI) Stop() { C.MPI_Finalize() } -// WorldRank returns the processor rank/ID within the World communicator -func WorldRank() (rank int) { +// WorldRank returns the rank of this process within the world communicator. +func (m *MPI) WorldRank() int { var r int32 C.MPI_Comm_rank(C.World, (*C.int)(unsafe.Pointer(&r))) return int(r) } -// WorldSize returns the number of processors in the World communicator -func WorldSize() (size int) { +// WorldSize returns the number of processes in the world communicator. +func (m *MPI) WorldSize() int { var s int32 C.MPI_Comm_size(C.World, (*C.int)(unsafe.Pointer(&s))) return int(s) } -func WorldTime() float64 { +// WorldTime returns the elapsed wall-clock time in seconds, as reported by +// MPI_Wtime. Useful for portable high-resolution timing. +func (m *MPI) WorldTime() float64 { return float64(C.MPI_Wtime()) } -// Communicator holds the World communicator or a subset communicator +// Communicator wraps an MPI communicator and its associated process group. +// Use NewCommunicator to obtain one. type Communicator struct { comm C.MPI_Comm group C.MPI_Group - MaxTag int + maxTag int } -// NewCommunicator creates a new communicator or returns the World communicator -// ranks -- World indices of processors in this Communicator. -// use nil or empty to get the World Communicator -func NewCommunicator(ranks []int) *Communicator { +// NewCommunicator creates a communicator containing the processes identified +// by ranks. If ranks is nil or empty, the world communicator is returned. +func (m *MPI) NewCommunicator(ranks []int) *Communicator { var o Communicator if len(ranks) == 0 { o.comm = C.World C.MPI_Comm_group(C.World, &o.group) - maxtag, err := o.GetMaxTag() + maxtag, err := o.MaxTag() if err != nil { panic(err) } - o.MaxTag = maxtag + o.maxTag = maxtag return &o } rs := make([]int32, len(ranks)) - for i := 0; i < len(ranks); i++ { + for i := range ranks { rs[i] = int32(ranks[i]) } n := C.int(len(ranks)) - r := (*C.int)(unsafe.Pointer(&rs[0])) + r := (*C.int)(unsafe.Pointer(unsafe.SliceData(rs))) var wgroup C.MPI_Group C.MPI_Comm_group(C.World, &wgroup) C.MPI_Group_incl(wgroup, n, r, &o.group) @@ -250,600 +254,103 @@ func NewCommunicator(ranks []int) *Communicator { // SplitType splits the communicator using MPI_Comm_split_type. // func (o *Communicator) SplitType(type int) -// Rank returns the processor rank/ID -func (o *Communicator) Rank() (rank int) { +// Rank returns the rank of this process within the communicator. +func (o *Communicator) Rank() int { var r int32 C.MPI_Comm_rank(o.comm, (*C.int)(unsafe.Pointer(&r))) return int(r) } -// Size returns the number of processors -func (o *Communicator) Size() (size int) { +// Size returns the number of processes in the communicator. +func (o *Communicator) Size() int { var s int32 C.MPI_Comm_size(o.comm, (*C.int)(unsafe.Pointer(&s))) return int(s) } -// Abort aborts MPI +// Abort terminates all processes in the communicator with the given error code. func (o *Communicator) Abort(errcode int) { C.MPI_Abort(o.comm, C.int(errcode)) } -// Barrier forces synchronisation +// Barrier blocks until all processes in the communicator have called Barrier. func (o *Communicator) Barrier() { C.MPI_Barrier(o.comm) } -// BcastBytes broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastBytes(x []byte, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Int], C.int(root), o.comm) -} - -// BcastUint32s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastUint32s(x []uint32, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Uint], C.int(root), o.comm) -} - -// BcastInt32s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastInt32s(x []int32, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Int], C.int(root), o.comm) -} - -// BcastUint64s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastUint64s(x []uint64, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Ulong], C.int(root), o.comm) -} - -// BcastInt64s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastInt64s(x []int64, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Long], C.int(root), o.comm) -} - -// BcastFloat32s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastFloat32s(x []float32, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Float], C.int(root), o.comm) -} - -// BcastFloat64s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastFloat64s(x []float64, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Double], C.int(root), o.comm) +// Bcast broadcasts x from the root process to all other processes in the +// communicator. All processes must call Bcast with the same root and a +// slice of the same length. +func (o *Communicator) Bcast[T goTypes](x []T, root int) { + C.MPI_Bcast(unsafe.Pointer(unsafe.SliceData(x)), C.int(len(x)), dataTypeOf[T](), C.int(root), o.comm) } -// BcastComplex128s broadcasts slice from root `root` to all other processors -func (o *Communicator) BcastComplex128s(x []complex128, root int) { - buf := unsafe.Pointer(&x[0]) - C.MPI_Bcast(buf, C.int(len(x)), dataTypes[Complex], C.int(root), o.comm) -} - -// ReduceBytes performs a distributed reduce operation on bytes, accumulating the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceBytes(dest, orig []byte, op Op, root int) error { - d := Byte - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) +// Reduce applies op to orig across all processes and writes the result +// into dest on the root process. dest and orig must be different slices. +// Returns an error if op is not valid for the data type. +func (o *Communicator) Reduce[T goTypes](dest, orig []T, op Op, root int) error { + c_datatype, valid_for_op := getDataTypeAndValidate[T](op) + if !valid_for_op { + return fmt.Errorf("DataType %T cannot be used with Operation %v", *new(T), op) } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) + C.MPI_Reduce(unsafe.Pointer(unsafe.SliceData(orig)), unsafe.Pointer(unsafe.SliceData(dest)), C.int(len(dest)), c_datatype, ops[op], C.int(root), o.comm) return nil } -// ReduceUint32s performs a distributed reduce operation on `uint32`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceUint32s(dest, orig []uint32, op Op, root int) error { - d := Uint - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) +// Allreduce applies op to orig across all processes and writes the result +// into dest on every process. dest and orig must be different slices. +// Returns an error if op is not valid for the data type. +func (o *Communicator) Allreduce[T goTypes](dest, orig []T, op Op) error { + c_datatype, valid_for_op := getDataTypeAndValidate[T](op) + if !valid_for_op { + return fmt.Errorf("DataType %T cannot be used with Operation %v", *new(T), op) } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) + C.MPI_Allreduce(unsafe.Pointer(unsafe.SliceData(orig)), unsafe.Pointer(unsafe.SliceData(dest)), C.int(len(dest)), c_datatype, ops[op], o.comm) return nil } -// ReduceInt32s performs a distributed reduce operation on `int32`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceInt32s(dest, orig []int32, op Op, root int) error { - d := Int - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil +// Send sends vals to processor toID with the given tag. +func (o *Communicator) Send[T goTypes](vals []T, toID int, tag int) { + C.MPI_Send(unsafe.Pointer(unsafe.SliceData(vals)), C.int(len(vals)), dataTypeOf[T](), C.int(toID), C.int(tag), o.comm) } -// ReduceUInt64s performs a distributed reduce operation on `uint64`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceUint64s(dest, orig []uint64, op Op, root int) error { - d := Ulong - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil -} - -// ReduceInt64s performs a distributed reduce operation on `int64`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceInt64s(dest, orig []int64, op Op, root int) error { - d := Long - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil -} - -// ReduceFloat32s performs a distributed reduce operation on `float32`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceFloat32s(dest, orig []float32, op Op, root int) error { - d := Float - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil -} - -// ReduceFloat64s performs a distributed reduce operation on `float64`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceFloat64s(dest, orig []float64, op Op, root int) error { - d := Double - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil -} - -// ReduceComplex128s performs a distributed reduce operation on `complex128`s, accumulating -// the operation on the given root. -// Note: dest and orig must be different slices. -func (o *Communicator) ReduceComplex128s(dest, orig []complex128, op Op, root int) error { - d := Complex - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Reduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], C.int(root), o.comm) - return nil -} - -// AllreduceBytes performs a distributed allreduce operation on bytes, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceBytes(dest, orig []byte, op Op, root int) error { - d := Byte - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceUint32s performs a distributed allreduce operation on `int32`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceUint32s(dest, orig []uint32, op Op, root int) error { - d := Uint - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceInt32s performs a distributed allreduce operation on `int32`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceInt32s(dest, orig []int32, op Op, root int) error { - d := Int - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceUint64s performs a distributed allreduce operation on `int64`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceUint64s(dest, orig []uint64, op Op, root int) error { - d := Ulong - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceInt64s performs a distributed allreduce operation on `int64`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceInt64s(dest, orig []int64, op Op, root int) error { - d := Long - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceFloat32s performs a distributed allreduce operation on `float32`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceFloat32s(dest, orig []float32, op Op, root int) error { - d := Float - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceFloat64s performs a distributed allreduce operation on `float64`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceFloat64s(dest, orig []float64, op Op, root int) error { - d := Double - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// AllreduceComplex128s performs a distributed allreduce operation on `complex128`s, accumulating -// the operation on all roots. -// Note: dest and orig must be different slices. -func (o *Communicator) AllreduceComplex128s(dest, orig []complex128, op Op, root int) error { - d := Complex - if !isValidDataTypeForOp(d, op) { - return fmt.Errorf("DataType %v cannot be used with Operation %v", d, op) - } - sendbuf := unsafe.Pointer(&orig[0]) - recvbuf := unsafe.Pointer(&dest[0]) - C.MPI_Allreduce(sendbuf, recvbuf, C.int(len(dest)), dataTypes[d], ops[op], o.comm) - return nil -} - -// SendBytes sends values to processor toID with given tag -func (o *Communicator) SendBytes(vals []byte, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Byte], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocBytes receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocBytes(vals []byte, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) +// RecvPrealloc receives into the preallocated slice vals from processor +// fromID with the given tag, and returns the resulting Status. +func (o *Communicator) RecvPrealloc[T goTypes](vals []T, fromID int, tag int) Status { status := Status{} - - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Byte], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) + C.MPI_Recv(unsafe.Pointer(unsafe.SliceData(vals)), C.int(len(vals)), dataTypeOf[T](), C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) return status } -// MrecvPreallocBytes receives values from processor fromId with given tag with threading -func (o *Communicator) MrecvPreallocBytes(vals []byte, fromID int, tag int, msg C.MPI_Message) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - - C.MPI_Mrecv(buf, C.int(len(vals)), dataTypes[Byte], &msg, &(status.mpiStatus)) - return status -} - -// RecvBytes returns a slice of bytes received from processor fromId with given tag. -func (o *Communicator) RecvBytes(fromID int, tag int) ([]byte, Status) { - l := o.Probe(fromID, tag).GetCount(Byte) - buf := make([]byte, l) - status := o.RecvPreallocBytes(buf, fromID, tag) - return buf, status -} - -// MrecvBytes returns a slice of bytes received from processor fromId with given tag. -func (o *Communicator) MrecvBytes(fromID int, tag int) ([]byte, Status) { - // runtime.LockOSThread() - pstatus, msg := o.Mprobe(fromID, tag) - l := pstatus.GetCount(Byte) - buf := make([]byte, l) - status := o.MrecvPreallocBytes(buf, fromID, tag, msg) - return buf, status -} - -// SendUint32s sends values to processor toID with given tag -func (o *Communicator) SendUInt32s(vals []uint32, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Uint], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocUint32s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocUint32s(vals []uint32, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Uint], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - - return status -} - -// RecvUint32s returns a slice of bytes received from processor fromId with given tag. -func (o *Communicator) RecvUint32s(fromID int, tag int) ([]uint32, Status) { - l := o.Probe(fromID, tag).GetCount(Uint) - buf := make([]uint32, l) - status := o.RecvPreallocUint32s(buf, fromID, tag) - return buf, status -} - -// SendInt32s sends values to processor toID with given tag -func (o *Communicator) SendInt32s(vals []int32, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Int], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocInt32s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocInt32s(vals []int32, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Int], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return status -} - -// RecvInt32s returns a slice of `int32`s received from processor fromId with given tag. -func (o *Communicator) RecvInt32s(fromID int, tag int) ([]int32, Status) { - l := o.Probe(fromID, tag).GetCount(Int) - buf := make([]int32, l) - status := o.RecvPreallocInt32s(buf, fromID, tag) - return buf, status -} - -// SendUint64s sends values to processor toID with given tag -func (o *Communicator) SendUint64s(vals []uint64, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Ulong], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocUint64s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocUint64s(vals []uint64, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Ulong], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return status -} - -// RecvUint64s returns a slice of `uint64`s received from processor fromId with given tag. -func (o *Communicator) RecvUint64s(fromID int, tag int) ([]uint64, Status) { - l := o.Probe(fromID, tag).GetCount(Ulong) - buf := make([]uint64, l) - status := o.RecvPreallocUint64s(buf, fromID, tag) - return buf, status -} - -// SendInt64s sends values to processor toID with given tag -func (o *Communicator) SendInt64s(vals []int64, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Long], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocInt64s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocInt64s(vals []int64, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Long], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return status -} - -// RecvInt64s returns a slice of `int64`s received from processor fromId with given tag. -func (o *Communicator) RecvInt64s(fromID int, tag int) ([]int64, Status) { - l := o.Probe(fromID, tag).GetCount(Long) - buf := make([]int64, l) - status := o.RecvPreallocInt64s(buf, fromID, tag) - return buf, status -} - -// SendFloat64s sends values to processor toID with given tag -func (o *Communicator) SendFloat64s(vals []float64, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Double], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocFloat64s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocFloat64s(vals []float64, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Double], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return status -} - -// RecvFloat64s returns a slice of `float64`s received from processor fromId with given tag. -func (o *Communicator) RecvFloat64s(fromID int, tag int) ([]float64, Status) { - l := o.Probe(fromID, tag).GetCount(Double) - buf := make([]float64, l) - status := o.RecvPreallocFloat64s(buf, fromID, tag) +// Recv allocates and returns a slice received from processor fromID +// with the given tag. +func (o *Communicator) Recv[T goTypes](fromID int, tag int) ([]T, Status) { + l := o.Probe(fromID, tag).Count[T]() + buf := make([]T, l) + status := o.RecvPrealloc(buf, fromID, tag) return buf, status } -// SendComplex128s sends values to processor toID with given tag -func (o *Communicator) SendComplex128s(vals []complex128, toID int, tag int) { - buf := unsafe.Pointer(&vals[0]) - C.MPI_Send(buf, C.int(len(vals)), dataTypes[Complex], C.int(toID), C.int(tag), o.comm) -} - -// RecvPreallocComplex128s receives values from processor fromId with given tag -func (o *Communicator) RecvPreallocComplex128s(vals []complex128, fromID int, tag int) Status { - buf := unsafe.Pointer(&vals[0]) - status := Status{} - C.MPI_Recv(buf, C.int(len(vals)), dataTypes[Complex], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return status -} - -// RecvComplex128s returns a slice of `complex128`s received from processor fromId with given tag. -func (o *Communicator) RecvComplex128s(fromID int, tag int) ([]complex128, Status) { - l := o.Probe(fromID, tag).GetCount(Complex) - buf := make([]complex128, l) - status := o.RecvPreallocComplex128s(buf, fromID, tag) - return buf, status -} - -////////////////////////////////////////////////////////////////////////////// -// SendByte sends one byte to processor toID with given tag -func (o *Communicator) SendByte(v byte, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Byte], C.int(toID), C.int(tag), o.comm) -} - -// RecvByte receives one byte from processor fromId with given tag -func (o *Communicator) RecvByte(fromID, tag int) (byte, Status) { - var v byte - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Byte], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return v, status -} - -// SendUint sends one `uint` to processor toID with given tag -func (o *Communicator) SendUint32(v uint32, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Uint], C.int(toID), C.int(tag), o.comm) +// SendOne sends a single value to processor toID with the given tag. +func (o *Communicator) SendOne[T goTypes](v T, toID int, tag int) { + C.MPI_Send(unsafe.Pointer(&v), 1, dataTypeOf[T](), C.int(toID), C.int(tag), o.comm) } -// RecvUint receives one `uint` from processor fromId with given tag -func (o *Communicator) RecvUint32(fromID, tag int) (uint32, Status) { - var v uint32 - buf := unsafe.Pointer(&v) +// RecvOne receives a single value from processor fromID with the given tag. +func (o *Communicator) RecvOne[T goTypes](fromID, tag int) (T, Status) { + var v T status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Uint], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) + C.MPI_Recv(unsafe.Pointer(&v), 1, dataTypeOf[T](), C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) return v, status } -// SendInt sends one `int` to processor toID with given tag -func (o *Communicator) SendInt32(v int32, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Int], C.int(toID), C.int(tag), o.comm) -} - -// RecvInt receives one `int` from processor fromId with given tag -func (o *Communicator) RecvInt32(fromID, tag int) (int32, Status) { - var v int32 - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Int], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return v, status -} - -// SendUint32 sends one `uint32` to processor toID with given tag -func (o *Communicator) SendUint64(v uint64, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Ulong], C.int(toID), C.int(tag), o.comm) -} - -// RecvUlong receives one `uint32` from processor fromId with given tag -func (o *Communicator) RecvUint64(fromID, tag int) (uint64, Status) { - var v uint64 - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Ulong], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return v, status -} - -// SendLong sends one `int64` to processor toID with given tag -func (o *Communicator) SendInt64(v int64, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Long], C.int(toID), C.int(tag), o.comm) -} - -// RecvLong receives one `int64` from processor fromId with given tag -func (o *Communicator) RecvInt64(fromID, tag int) (int64, Status) { - var v int64 - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Long], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return v, status -} - -// SendDouble sends one `float64` to processor toID with given tag -func (o *Communicator) SendFloat64(v float64, toID int, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Double], C.int(toID), C.int(tag), o.comm) -} - -// RecvDouble receives one `float64` from processor fromId with given tag -func (o *Communicator) RecvFloat64(fromID, tag int) (float64, Status) { - var v float64 - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Double], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - return v, status -} - -// SendComplex128 sends one `complex128` to processor toID (integer version) -func (o *Communicator) SendComplex128(v complex128, toID, tag int) { - buf := unsafe.Pointer(&v) - C.MPI_Send(buf, 1, dataTypes[Complex], C.int(toID), C.int(tag), o.comm) -} - -// RecvComplex128 receives one `complex128` from processor fromId -func (o *Communicator) RecvComplex128(fromID, tag int) (complex128, Status) { - var v complex128 - buf := unsafe.Pointer(&v) - status := Status{} - C.MPI_Recv(buf, 1, dataTypes[Complex], C.int(fromID), C.int(tag), o.comm, &(status.mpiStatus)) - - return v, status -} - -// SendString is a convenience function to send one string to processor toID with given tag. -func (o *Communicator) SendString(s string, toID, tag int) { - o.SendBytes([]byte(s), toID, tag) -} - -// RecvString is a convenience function to receive a string from processor fromId with given tag. -func (o *Communicator) RecvString(fromID, tag int) (string, Status) { - recv_bytes, status := o.RecvBytes(fromID, tag) - return string(recv_bytes), status -} - -// IProbe will return a boolean indicating whether a message is -// waiting from a source with a given tag, and a status structure. -func (o *Communicator) Iprobe(source, tag int) (bool, *Status) { +// Iprobe reports whether a message from source with the given tag is available +// without blocking. It returns true and the message Status if a message is +// waiting, or false and a zero Status if not. +func (o *Communicator) Iprobe(source, tag int) (bool, Status) { var s Status var b C.int C.MPI_Iprobe(C.int(source), C.int(tag), o.comm, &b, &(s.mpiStatus)) - return b == 1, &s + return b == 1, s } diff --git a/mpi_test.go b/mpi_test.go index a00c4f8..ad2dbc8 100644 --- a/mpi_test.go +++ b/mpi_test.go @@ -9,957 +9,367 @@ import ( "testing" ) -const tol = 1e-10 const ( - MaxUint32 = ^uint32(0) - MaxUint64 = ^uint64(0) + tol = 1e-10 + tolF32 = 1e-5 ) -func setSliceByte(x []byte, rank int, offset byte) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = byte(rank+1) + offset - } else { - x[i] = 0xff - } - } -} -func setSliceUint32(x []uint32, rank int, offset uint32) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = uint32(rank+1) + offset - } else { - x[i] = MaxUint32 - } - } +var opNames = [len(ops)]string{ + "sum", "min", "max", "prod", + "land", "lor", "lxor", + "band", "bor", "bxor", } -func setSliceInt32(x []int32, rank int, offset int32) { - for i := 0; i < len(x); i++ { +// setSlice sets x[rank] = rankVal and all other elements to fillVal. +func setSlice[T goTypes](x []T, rank int, rankVal, fillVal T) { + for i := range x { if i == rank { - x[i] = int32(rank+1) + offset + x[i] = rankVal } else { - x[i] = -1 + x[i] = fillVal } } } -func setSliceUint64(x []uint64, rank int, offset uint64) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = uint64(rank+1) + offset - } else { - x[i] = MaxUint64 - } +// valEqual compares two values with type-appropriate equality. +func valEqual[T goTypes](a, b T) bool { + switch x := any(a).(type) { + case float32: + return math.Abs(float64(x)-float64(any(b).(float32))) <= tolF32 + case float64: + return math.Abs(x-any(b).(float64)) <= tol + case complex64: + bv := any(b).(complex64) + return math.Abs(float64(real(x))-float64(real(bv))) <= tolF32 && + math.Abs(float64(imag(x))-float64(imag(bv))) <= tolF32 + case complex128: + bv := any(b).(complex128) + return math.Abs(real(x)-real(bv)) <= tol && + math.Abs(imag(x)-imag(bv)) <= tol + default: + return a == b } } -func setSliceInt64(x []int64, rank int, offset int64) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = int64(rank+1) + offset - } else { - x[i] = -1 - } - } -} - -func setSliceFloat32(x []float32, rank int, offset float32) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = float32(rank+1) + offset - } else { - x[i] = -1 - } - } -} -func setSliceFloat64(x []float64, rank int, offset float64) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = float64(rank+1) + offset - } else { - x[i] = -1 - } - } -} - -func setSliceComplex128(x []complex128, rank int, offset complex128) { - for i := 0; i < len(x); i++ { - if i == rank { - x[i] = complex(float64(rank+1), float64(rank+1)/10.0) + offset - } else { - x[i] = complex(float64(-1), float64(-1)) - } - } -} - -func chkStatus(s Status, source, tag int) bool { - return s.GetSource() == source && s.GetTag() == tag -} - -func chkArraysEqualByte(a, b []byte) bool { +// slicesEqual compares two slices element-wise using valEqual. +func slicesEqual[T goTypes](a, b []T) bool { if len(a) != len(b) { return false } for i := range a { - if a[i] != b[i] { + if !valEqual(a[i], b[i]) { return false } } return true } -func chkArraysEqualUint32(a, b []uint32) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if a[i] != b[i] { - return false - } - } - return true +func chkStatus(s Status, source, tag int) bool { + return s.Source() == source && s.Tag() == tag } -func chkArraysEqualInt32(a, b []int32) bool { - if len(a) != len(b) { - return false +// testBcast verifies Bcast distributes {1,2,3,4} from root to all ranks. +func testBcast[T goTypes](t *testing.T, A *Communicator, root int) { + t.Helper() + b := make([]T, 4) + exp := []T{T(1), T(2), T(3), T(4)} + if A.Rank() == root { + copy(b, exp) } - for i := range a { - if a[i] != b[i] { - return false - } + A.Bcast(b, root) + if !slicesEqual(b, exp) { + t.Errorf("got %v, want %v", b, exp) } - return true } -func chkArraysEqualUint64(a, b []uint64) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if a[i] != b[i] { - return false +// testReduceOp runs a single reduce op and checks the result at root. +// A nil or empty expRoot means the op is invalid or not checked for this type. +func testReduceOp[T goTypes](t *testing.T, A *Communicator, op Op, root int, x, expRoot []T) { + t.Helper() + res := make([]T, len(x)) + err := A.Reduce(res, x, op, root) + valid := isValidDataTypeForOp[T](op) + if err != nil { + if valid { + t.Errorf("unexpected error for valid op: %v", err) } + return } - return true -} - -func chkArraysEqualInt64(a, b []int64) bool { - if len(a) != len(b) { - return false + if !valid { + t.Errorf("no error for invalid op") + return } - for i := range a { - if a[i] != b[i] { - return false - } + if A.Rank() == root && len(expRoot) > 0 && !slicesEqual(res, expRoot) { + t.Errorf("got %v, want %v", res, expRoot) } - return true } -func chkArraysEqualFloat32(a, b []float32) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if math.Abs(float64(a[i])-float64(b[i])) > tol { - return false +// testAllreduceOp runs a single allreduce op and checks the result on all ranks. +// A nil or empty exp means the op is invalid or not checked for this type. +func testAllreduceOp[T goTypes](t *testing.T, A *Communicator, op Op, x, exp []T) { + t.Helper() + res := make([]T, len(x)) + err := A.Allreduce(res, x, op) + valid := isValidDataTypeForOp[T](op) + if err != nil { + if valid { + t.Errorf("unexpected error for valid op: %v", err) } + return + } + if !valid { + t.Errorf("no error for invalid op") + return + } + if len(exp) > 0 && !slicesEqual(res, exp) { + t.Errorf("got %v, want %v", res, exp) } - return true } -func chkArraysEqualFloat64(a, b []float64) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if math.Abs(a[i]-b[i]) > tol { - return false - } +// signedResults returns expected reduce/allreduce results for signed integer types. +// The pattern is identical across int8, int16, int32, int64 since the test values +// (-1 fill, rank+1 for rank slot) produce the same relative results. +func signedResults[T interface{ int8 | int16 | int32 | int64 }]() [10][]T { + return [10][]T{ + {-2, -1, 0, 1}, // OpSum + {-1, -1, -1, -1}, // OpMin + {1, 2, 3, 4}, // OpMax + {-1, -2, -3, -4}, // OpProd + {1, 1, 1, 1}, // OpLand + {1, 1, 1, 1}, // OpLor + {0, 0, 0, 0}, // OpLxor + {1, 2, 3, 4}, // OpBand + {-1, -1, -1, -1}, // OpBor + {-2, -3, -4, -5}, // OpBxor } - return true } -func chkArraysEqualComplex128(a, b []complex128) bool { - if len(a) != len(b) { - return false +// unsignedResults returns expected reduce/allreduce results for unsigned integer types. +// Uses MaxType (^T(0)) as the fill value. +func unsignedResults[T interface{ byte | uint16 | uint32 | uint64 }]() [10][]T { + max := ^T(0) + return [10][]T{ + {max - 1, max, 0, 1}, // OpSum + {1, 2, 3, 4}, // OpMin + {max, max, max, max}, // OpMax + {max, max - 1, max - 2, max - 3}, // OpProd + {1, 1, 1, 1}, // OpLand + {1, 1, 1, 1}, // OpLor + {0, 0, 0, 0}, // OpLxor + {1, 2, 3, 4}, // OpBand + {max, max, max, max}, // OpBor + {max - 1, max - 2, max - 3, max - 4}, // OpBxor } - for i := range a { - if math.Abs(real(a[i])-real(b[i])) > tol || math.Abs(imag(a[i])-imag(b[i])) > tol { - return false - } +} + +// floatResults returns expected reduce/allreduce results for float types. +// Logical and bitwise ops are invalid for floats; their entries are nil. +func floatResults[T interface{ float32 | float64 }]() [10][]T { + return [10][]T{ + {-2, -1, 0, 1}, // OpSum + {-1, -1, -1, -1}, // OpMin + {1, 2, 3, 4}, // OpMax + {-1, -2, -3, -4}, // OpProd + nil, nil, nil, nil, nil, nil, // OpLand–OpBxor invalid } - return true } func bcast(A *Communicator) func(*testing.T) { return func(t *testing.T) { root := 3 - t.Run("byte", func(t *testing.T) { - b := make([]byte, 4) - if A.Rank() == root { - for i := range b { - b[i] = byte(1 + i) - } - } - var exp = []byte{1, 2, 3, 4} - A.BcastBytes(b, root) - if !chkArraysEqualByte(b, exp) { - t.Errorf("received %v, expected %v", b, exp) - } - }) - A.Barrier() - - t.Run("uint32", func(t *testing.T) { - u32 := make([]uint32, 4) - if A.Rank() == root { - for i := 0; i < len(u32); i++ { - u32[i] = uint32(1 + i) - } - } - var exp = []uint32{1, 2, 3, 4} - A.BcastUint32s(u32, root) - if !chkArraysEqualUint32(u32, exp) { - t.Errorf("received %v, expected %v", u32, exp) - } - }) - A.Barrier() - - t.Run("int32", func(t *testing.T) { - i32 := make([]int32, 4) - if A.Rank() == root { - for i := 0; i < len(i32); i++ { - i32[i] = int32(1 + i) - } - } - var exp = []int32{1, 2, 3, 4} - A.BcastInt32s(i32, root) - if !chkArraysEqualInt32(i32, exp) { - t.Errorf("received %v, expected %v", i32, exp) - } - - }) - A.Barrier() - - t.Run("uint64", func(t *testing.T) { - u64 := make([]uint64, 4) - if A.Rank() == root { - for i := 0; i < len(u64); i++ { - u64[i] = uint64(1 + i) - } - } - exp := []uint64{1, 2, 3, 4} - A.BcastUint64s(u64, root) - if !chkArraysEqualUint64(u64, exp) { - t.Errorf("received %v, expected %v", u64, exp) - } - }) - A.Barrier() - - t.Run("int64", func(t *testing.T) { - i64 := make([]int64, 4) - if A.Rank() == root { - for i := 0; i < len(i64); i++ { - i64[i] = int64(1 + i) - } - } - var exp = []int64{1, 2, 3, 4} - A.BcastInt64s(i64, root) - if !chkArraysEqualInt64(i64, exp) { - t.Errorf("received %v, expected %v", i64, exp) - } - }) - A.Barrier() - - t.Run("float32", func(t *testing.T) { - f32 := make([]float32, 4) - if A.Rank() == root { - for i := 0; i < len(f32); i++ { - f32[i] = float32(1 + i) - } - } - var exp = []float32{1, 2, 3, 4} - A.BcastFloat32s(f32, root) - if !chkArraysEqualFloat32(f32, exp) { - t.Errorf("received %v, expected %v", f32, exp) - } - }) - A.Barrier() - - t.Run("float64", func(t *testing.T) { - f64 := make([]float64, 4) - if A.Rank() == root { - for i := 0; i < len(f64); i++ { - f64[i] = float64(1 + i) - } - } - var exp = []float64{1, 2, 3, 4} - A.BcastFloat64s(f64, root) - if !chkArraysEqualFloat64(f64, exp) { - t.Errorf("received %v, expected %v", f64, exp) - } - }) - A.Barrier() - - t.Run("complex128", func(t *testing.T) { - c128 := make([]complex128, 4) - if A.Rank() == root { - for i := 0; i < len(c128); i++ { - c128[i] = complex(float64(1+i), float64(i)) - } - } - var exp = []complex128{complex(1, 0), complex(2, 1), complex(3, 2), complex(4, 3)} - A.BcastComplex128s(c128, root) - if !chkArraysEqualComplex128(c128, exp) { - t.Errorf("received %v, expected %v", c128, exp) - } - }) - A.Barrier() + for _, tc := range []struct { + name string + run func() + }{ + {"int8", func() { testBcast[int8](t, A, root) }}, + {"byte", func() { testBcast[byte](t, A, root) }}, + {"int16", func() { testBcast[int16](t, A, root) }}, + {"uint16", func() { testBcast[uint16](t, A, root) }}, + {"int32", func() { testBcast[int32](t, A, root) }}, + {"uint32", func() { testBcast[uint32](t, A, root) }}, + {"int64", func() { testBcast[int64](t, A, root) }}, + {"uint64", func() { testBcast[uint64](t, A, root) }}, + {"float32", func() { testBcast[float32](t, A, root) }}, + {"float64", func() { testBcast[float64](t, A, root) }}, + {"complex64", func() { testBcast[complex64](t, A, root) }}, + {"complex128", func() { testBcast[complex128](t, A, root) }}, + } { + t.Run(tc.name, func(*testing.T) { tc.run() }) + A.Barrier() + } } } func reduce(A *Communicator) func(*testing.T) { root := 3 - testNames := [...]string{ - "sum", - "min", - "max", - "prod", - "land", - "lor", - "lxor", - "band", - "bor", - "bxor", - } - return func(t *testing.T) { - for opidx := range ops { - t.Run(testNames[opidx], func(t *testing.T) { - op := Op(opidx) - t.Run("byte", func(t *testing.T) { - results := [...][]byte{ - {0xfe, 0xff, 0, 1}, // sum - {1, 2, 3, 4}, // min - {0xff, 0xff, 0xff, 0xff}, // max - {0xff, 0xfe, 0xfd, 0xfc}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {0xff, 0xff, 0xff, 0xff}, // bor - {0xfe, 0xfd, 0xfc, 0xfb}, // bxor - } - - x := make([]byte, 4) - setSliceByte(x, int(A.Rank()), 0) - res := make([]byte, len(x)) - err := A.ReduceBytes(res, x, op, root) - valid := isValidDataTypeForOp(Byte, op) - - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]byte, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualByte(res, exp) { - fmt.Println("x = ", x) - t.Errorf("rank %d received %v, expected %v", A.Rank(), res, exp) - } - }) - A.Barrier() - - t.Run("uint32", func(t *testing.T) { - results := [...][]uint32{ - {MaxUint32 - 1, MaxUint32, 0, 1}, // sum - {1, 2, 3, 4}, // min - {MaxUint32, MaxUint32, MaxUint32, MaxUint32}, // max - {MaxUint32, MaxUint32 - 1, MaxUint32 - 2, MaxUint32 - 3}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {MaxUint32, MaxUint32, MaxUint32, MaxUint32}, // bor - {MaxUint32 - 1, MaxUint32 - 2, MaxUint32 - 3, MaxUint32 - 4}, // bxor - } - - x := make([]uint32, 4) - setSliceUint32(x, int(A.Rank()), 0) - res := make([]uint32, len(x)) - err := A.ReduceUint32s(res, x, op, root) - valid := isValidDataTypeForOp(Uint, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]uint32, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualUint32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("int32", func(t *testing.T) { - results := [...][]int32{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {-1, -1, -1, -1}, // bor - {-2, -3, -4, -5}, // bxor - } - x := make([]int32, 4) - setSliceInt32(x, int(A.Rank()), 0) - res := make([]int32, len(x)) - err := A.ReduceInt32s(res, x, op, root) - valid := isValidDataTypeForOp(Int, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]int32, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualInt32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("uint64", func(t *testing.T) { - results := [...][]uint64{ - {MaxUint64 - 1, MaxUint64, 0, 1}, // sum - {1, 2, 3, 4}, // min - {MaxUint64, MaxUint64, MaxUint64, MaxUint64}, // max - {MaxUint64, MaxUint64 - 1, MaxUint64 - 2, MaxUint64 - 3}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {MaxUint64, MaxUint64, MaxUint64, MaxUint64}, // bor - {MaxUint64 - 1, MaxUint64 - 2, MaxUint64 - 3, MaxUint64 - 4}, // bxor - } - - x := make([]uint64, 4) - setSliceUint64(x, int(A.Rank()), 0) - res := make([]uint64, len(x)) - err := A.ReduceUint64s(res, x, op, root) - valid := isValidDataTypeForOp(Ulong, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]uint64, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualUint64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("int64", func(t *testing.T) { - results := [...][]int64{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {-1, -1, -1, -1}, // bor - {-2, -3, -4, -5}, // bxor - } - x := make([]int64, 4) - setSliceInt64(x, int(A.Rank()), 0) - res := make([]int64, len(x)) - err := A.ReduceInt64s(res, x, op, root) - valid := isValidDataTypeForOp(Long, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]int64, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualInt64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("float32", func(t *testing.T) { - results := [...][]float32{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]float32, 4) - setSliceFloat32(x, int(A.Rank()), 0) - res := make([]float32, len(x)) - err := A.ReduceFloat32s(res, x, op, root) - valid := isValidDataTypeForOp(Float, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]float32, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualFloat32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("float64", func(t *testing.T) { - results := [...][]float64{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]float64, 4) - setSliceFloat64(x, int(A.Rank()), 0) - res := make([]float64, len(x)) - err := A.ReduceFloat64s(res, x, op, root) - valid := isValidDataTypeForOp(Double, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]float64, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualFloat64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("complex128", func(t *testing.T) { - results := [...][]complex128{ - {(-2 - 2.9i), (-1 - 2.8i), (0 - 2.7i), (1 - 2.6i)}, // sum - {}, // min - not tested - {}, // max - not tested + rank := int(A.Rank()) + + xi8 := make([]int8, 4) + setSlice(xi8, rank, int8(rank+1), int8(-1)) + xb := make([]byte, 4) + setSlice(xb, rank, byte(rank+1), ^byte(0)) + xi16 := make([]int16, 4) + setSlice(xi16, rank, int16(rank+1), int16(-1)) + xu16 := make([]uint16, 4) + setSlice(xu16, rank, uint16(rank+1), ^uint16(0)) + xi32 := make([]int32, 4) + setSlice(xi32, rank, int32(rank+1), int32(-1)) + xu32 := make([]uint32, 4) + setSlice(xu32, rank, uint32(rank+1), ^uint32(0)) + xi64 := make([]int64, 4) + setSlice(xi64, rank, int64(rank+1), int64(-1)) + xu64 := make([]uint64, 4) + setSlice(xu64, rank, uint64(rank+1), ^uint64(0)) + xf32 := make([]float32, 4) + setSlice(xf32, rank, float32(rank+1), float32(-1)) + xf64 := make([]float64, 4) + setSlice(xf64, rank, float64(rank+1), float64(-1)) + xc64 := make([]complex64, 4) + setSlice(xc64, rank, complex(float32(rank+1), float32(rank+1)/10), complex64(-1-1i)) + xc128 := make([]complex128, 4) + setSlice(xc128, rank, complex(float64(rank+1), float64(rank+1)/10), complex128(-1-1i)) + + ri8 := signedResults[int8]() + rb := unsignedResults[byte]() + ri16 := signedResults[int16]() + ru16 := unsignedResults[uint16]() + ri32 := signedResults[int32]() + ru32 := unsignedResults[uint32]() + ri64 := signedResults[int64]() + ru64 := unsignedResults[uint64]() + rf32 := floatResults[float32]() + rf64 := floatResults[float64]() + rc64 := [10][]complex64{ + {-2 - 2.9i, -1 - 2.8i, -2.7i, 1 - 2.6i}, + nil, nil, + {2.2 - 1.8i, 4.4 - 3.6i, 6.6 - 5.4i, 8.8 - 7.2i}, + nil, nil, nil, nil, nil, nil, + } + rc128 := [10][]complex128{ + {-2 - 2.9i, -1 - 2.8i, -2.7i, 1 - 2.6i}, + nil, nil, + {2.2 - 1.8i, 4.4 - 3.6i, 6.6 - 5.4i, 8.8 - 7.2i}, + nil, nil, nil, nil, nil, nil, + } - {(2.2 - 1.8i), (4.4 - 3.6i), (6.6 - 5.4i), (8.8 - 7.199999999999999i)}, //prod - {}, // prod - not tested - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]complex128, 4) - setSliceComplex128(x, int(A.Rank()), 0) - res := make([]complex128, len(x)) - err := A.ReduceComplex128s(res, x, op, root) - valid := isValidDataTypeForOp(Complex, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := make([]complex128, 4) - if A.Rank() == root { - exp = results[opidx] - } - if !chkArraysEqualComplex128(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() + for opidx := range ops { + op := Op(opidx) + t.Run(opNames[opidx], func(t *testing.T) { + for _, tc := range []struct { + name string + run func() + }{ + {"int8", func() { testReduceOp(t, A, op, root, xi8, ri8[opidx]) }}, + {"byte", func() { testReduceOp(t, A, op, root, xb, rb[opidx]) }}, + {"int16", func() { testReduceOp(t, A, op, root, xi16, ri16[opidx]) }}, + {"uint16", func() { testReduceOp(t, A, op, root, xu16, ru16[opidx]) }}, + {"int32", func() { testReduceOp(t, A, op, root, xi32, ri32[opidx]) }}, + {"uint32", func() { testReduceOp(t, A, op, root, xu32, ru32[opidx]) }}, + {"int64", func() { testReduceOp(t, A, op, root, xi64, ri64[opidx]) }}, + {"uint64", func() { testReduceOp(t, A, op, root, xu64, ru64[opidx]) }}, + {"float32", func() { testReduceOp(t, A, op, root, xf32, rf32[opidx]) }}, + {"float64", func() { testReduceOp(t, A, op, root, xf64, rf64[opidx]) }}, + {"complex64", func() { testReduceOp(t, A, op, root, xc64, rc64[opidx]) }}, + {"complex128", func() { testReduceOp(t, A, op, root, xc128, rc128[opidx]) }}, + } { + t.Run(tc.name, func(*testing.T) { tc.run() }) + A.Barrier() + } }) } } } -func allreduce(A *Communicator) func(*testing.T) { - root := 3 - testNames := [...]string{ - "sum", - "min", - "max", - "prod", - "land", - "lor", - "lxor", - "band", - "bor", - "bxor", - } +func allreduce(A *Communicator) func(*testing.T) { return func(t *testing.T) { - for opidx := range ops { - t.Run(testNames[opidx], func(t *testing.T) { - op := Op(opidx) - t.Run("byte", func(t *testing.T) { - results := [...][]byte{ - {0xfe, 0xff, 0, 1}, // sum - {1, 2, 3, 4}, // min - {0xff, 0xff, 0xff, 0xff}, // max - {0xff, 0xfe, 0xfd, 0xfc}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {0xff, 0xff, 0xff, 0xff}, // bor - {0xfe, 0xfd, 0xfc, 0xfb}, // bxor - } - - x := make([]byte, 4) - setSliceByte(x, int(A.Rank()), 0) - res := make([]byte, len(x)) - err := A.AllreduceBytes(res, x, op, root) - valid := isValidDataTypeForOp(Byte, op) - - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualByte(res, exp) { - fmt.Println("x = ", x) - t.Errorf("rank %d received %v, expected %v", A.Rank(), res, exp) - } - }) - A.Barrier() - - t.Run("uint32", func(t *testing.T) { - results := [...][]uint32{ - {MaxUint32 - 1, MaxUint32, 0, 1}, // sum - {1, 2, 3, 4}, // min - {MaxUint32, MaxUint32, MaxUint32, MaxUint32}, // max - {MaxUint32, MaxUint32 - 1, MaxUint32 - 2, MaxUint32 - 3}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {MaxUint32, MaxUint32, MaxUint32, MaxUint32}, // bor - {MaxUint32 - 1, MaxUint32 - 2, MaxUint32 - 3, MaxUint32 - 4}, // bxor - } - - x := make([]uint32, 4) - setSliceUint32(x, int(A.Rank()), 0) - res := make([]uint32, len(x)) - err := A.AllreduceUint32s(res, x, op, root) - valid := isValidDataTypeForOp(Uint, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualUint32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("int32", func(t *testing.T) { - results := [...][]int32{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {-1, -1, -1, -1}, // bor - {-2, -3, -4, -5}, // bxor - } - x := make([]int32, 4) - setSliceInt32(x, int(A.Rank()), 0) - res := make([]int32, len(x)) - err := A.AllreduceInt32s(res, x, op, root) - valid := isValidDataTypeForOp(Int, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualInt32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("uint64", func(t *testing.T) { - results := [...][]uint64{ - {MaxUint64 - 1, MaxUint64, 0, 1}, // sum - {1, 2, 3, 4}, // min - {MaxUint64, MaxUint64, MaxUint64, MaxUint64}, // max - {MaxUint64, MaxUint64 - 1, MaxUint64 - 2, MaxUint64 - 3}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {MaxUint64, MaxUint64, MaxUint64, MaxUint64}, // bor - {MaxUint64 - 1, MaxUint64 - 2, MaxUint64 - 3, MaxUint64 - 4}, // bxor - } - - x := make([]uint64, 4) - setSliceUint64(x, int(A.Rank()), 0) - res := make([]uint64, len(x)) - err := A.AllreduceUint64s(res, x, op, root) - valid := isValidDataTypeForOp(Ulong, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualUint64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("int64", func(t *testing.T) { - results := [...][]int64{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {1, 1, 1, 1}, // land - {1, 1, 1, 1}, // lor - {0, 0, 0, 0}, // lxor - {1, 2, 3, 4}, // band - {-1, -1, -1, -1}, // bor - {-2, -3, -4, -5}, // bxor - } - x := make([]int64, 4) - setSliceInt64(x, int(A.Rank()), 0) - res := make([]int64, len(x)) - err := A.AllreduceInt64s(res, x, op, root) - valid := isValidDataTypeForOp(Long, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualInt64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("float32", func(t *testing.T) { - results := [...][]float32{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]float32, 4) - setSliceFloat32(x, int(A.Rank()), 0) - res := make([]float32, len(x)) - err := A.AllreduceFloat32s(res, x, op, root) - valid := isValidDataTypeForOp(Float, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualFloat32(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("float64", func(t *testing.T) { - results := [...][]float64{ - {-2, -1, 0, 1}, // sum - {-1, -1, -1, -1}, // min - {1, 2, 3, 4}, // max - {-1, -2, -3, -4}, // prod - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]float64, 4) - setSliceFloat64(x, int(A.Rank()), 0) - res := make([]float64, len(x)) - err := A.AllreduceFloat64s(res, x, op, root) - valid := isValidDataTypeForOp(Double, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualFloat64(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() - - t.Run("complex128", func(t *testing.T) { - results := [...][]complex128{ - {(-2 - 2.9i), (-1 - 2.8i), (0 - 2.7i), (1 - 2.6i)}, // sum - {}, // min - not tested - {}, // max - not tested + rank := int(A.Rank()) + + xi8 := make([]int8, 4) + setSlice(xi8, rank, int8(rank+1), int8(-1)) + xb := make([]byte, 4) + setSlice(xb, rank, byte(rank+1), ^byte(0)) + xi16 := make([]int16, 4) + setSlice(xi16, rank, int16(rank+1), int16(-1)) + xu16 := make([]uint16, 4) + setSlice(xu16, rank, uint16(rank+1), ^uint16(0)) + xi32 := make([]int32, 4) + setSlice(xi32, rank, int32(rank+1), int32(-1)) + xu32 := make([]uint32, 4) + setSlice(xu32, rank, uint32(rank+1), ^uint32(0)) + xi64 := make([]int64, 4) + setSlice(xi64, rank, int64(rank+1), int64(-1)) + xu64 := make([]uint64, 4) + setSlice(xu64, rank, uint64(rank+1), ^uint64(0)) + xf32 := make([]float32, 4) + setSlice(xf32, rank, float32(rank+1), float32(-1)) + xf64 := make([]float64, 4) + setSlice(xf64, rank, float64(rank+1), float64(-1)) + xc64 := make([]complex64, 4) + setSlice(xc64, rank, complex(float32(rank+1), float32(rank+1)/10), complex64(-1-1i)) + xc128 := make([]complex128, 4) + setSlice(xc128, rank, complex(float64(rank+1), float64(rank+1)/10), complex128(-1-1i)) + + ri8 := signedResults[int8]() + rb := unsignedResults[byte]() + ri16 := signedResults[int16]() + ru16 := unsignedResults[uint16]() + ri32 := signedResults[int32]() + ru32 := unsignedResults[uint32]() + ri64 := signedResults[int64]() + ru64 := unsignedResults[uint64]() + rf32 := floatResults[float32]() + rf64 := floatResults[float64]() + rc64 := [10][]complex64{ + {-2 - 2.9i, -1 - 2.8i, -2.7i, 1 - 2.6i}, + nil, nil, + {2.2 - 1.8i, 4.4 - 3.6i, 6.6 - 5.4i, 8.8 - 7.2i}, + nil, nil, nil, nil, nil, nil, + } + rc128 := [10][]complex128{ + {-2 - 2.9i, -1 - 2.8i, -2.7i, 1 - 2.6i}, + nil, nil, + {2.2 - 1.8i, 4.4 - 3.6i, 6.6 - 5.4i, 8.8 - 7.2i}, + nil, nil, nil, nil, nil, nil, + } - {(2.2 - 1.8i), (4.4 - 3.6i), (6.6 - 5.4i), (8.8 - 7.199999999999999i)}, //prod - {}, // prod - not tested - {}, // land - not tested - {}, // lor - not tested - {}, // lxor - not tested - {}, // band - not tested - {}, // bor - not tested - {}, // bxor - not tested - } - x := make([]complex128, 4) - setSliceComplex128(x, int(A.Rank()), 0) - res := make([]complex128, len(x)) - err := A.AllreduceComplex128s(res, x, op, root) - valid := isValidDataTypeForOp(Complex, op) - if err != nil { - if valid { - t.Errorf("Improper error was thrown: valid data type for the op was supplied") - } - return - } - if !valid { - t.Errorf("Error should have been thrown: invalid data type for op not properly detected.") - } - exp := results[opidx] - if !chkArraysEqualComplex128(res, exp) { - t.Errorf("received %v, expected %v", res, exp) - } - }) - A.Barrier() + for opidx := range ops { + op := Op(opidx) + t.Run(opNames[opidx], func(t *testing.T) { + for _, tc := range []struct { + name string + run func() + }{ + {"int8", func() { testAllreduceOp(t, A, op, xi8, ri8[opidx]) }}, + {"byte", func() { testAllreduceOp(t, A, op, xb, rb[opidx]) }}, + {"int16", func() { testAllreduceOp(t, A, op, xi16, ri16[opidx]) }}, + {"uint16", func() { testAllreduceOp(t, A, op, xu16, ru16[opidx]) }}, + {"int32", func() { testAllreduceOp(t, A, op, xi32, ri32[opidx]) }}, + {"uint32", func() { testAllreduceOp(t, A, op, xu32, ru32[opidx]) }}, + {"int64", func() { testAllreduceOp(t, A, op, xi64, ri64[opidx]) }}, + {"uint64", func() { testAllreduceOp(t, A, op, xu64, ru64[opidx]) }}, + {"float32", func() { testAllreduceOp(t, A, op, xf32, rf32[opidx]) }}, + {"float64", func() { testAllreduceOp(t, A, op, xf64, rf64[opidx]) }}, + {"complex64", func() { testAllreduceOp(t, A, op, xc64, rc64[opidx]) }}, + {"complex128", func() { testAllreduceOp(t, A, op, xc128, rc128[opidx]) }}, + } { + t.Run(tc.name, func(*testing.T) { tc.run() }) + A.Barrier() + } }) } } } func TestMPI(t *testing.T) { - Start(true) - defer Stop() - if WorldSize() < 4 { + m, _ := Start() + defer m.Stop() + if m.WorldSize() < 4 { t.Fatal("These tests require 4 processors (are you running with mpirun?)\n") } - // subsets of processors - A := NewCommunicator([]int{0, 1, 2, 3}) - // if A.Rank() != 0 { - // os.Stdout, _ = os.Open(os.DevNull) - // } - // B := NewCommunicator([]int{0, 1, 2, 3}) + A := m.NewCommunicator([]int{0, 1, 2, 3}) t.Run("Bcast", bcast(A)) A.Barrier() @@ -967,63 +377,21 @@ func TestMPI(t *testing.T) { A.Barrier() t.Run("Allreduce", allreduce(A)) A.Barrier() - // t.Run("ReduceSumFloat64s", func(t *testing.T) { - // root := 3 - // x := make([]float64, 4) - // setSliceFloat64(x, int(A.Rank()), 0) - // res := make([]float64, len(x)) - // A.ReduceSumFloat64s(res, x, root) - // if A.Rank() == root { - // chkArraysEqualFloat64(t, res, []float64{-2, -1, 0, 1}) - // } else { - // chkArraysEqualFloat64(t, res, []float64{0, 0, 0, 0}) - // } - // }) - // A.Barrier() - // - // // AllReduceSum - // t.Run("AllReduceSumFloat64s", func(t *testing.T) { - // x := make([]float64, 4) - // res := make([]float64, 4) - // setSliceFloat64(x, int(A.Rank()), 0) - // A.AllReduceSumFloat64s(res, x) - // chkArraysEqualFloat64(t, res, []float64{-2, -1, 0, 1}) - // }) - // A.Barrier() - // - // // AllReduceMin - // t.Run("AllReduceMinFloat64s", func(t *testing.T) { - // x := make([]float64, 4) - // setSliceFloat64(x, int(A.Rank()), -3.5) - // res := make([]float64, len(x)) - // A.AllReduceMinFloat64s(res, x) - // // fmt.Println("allreducemin: rank", A.Rank(), "res = ", res) - // // fmt.Println("allreduceminL rank", A.Rank(), " x = ", x) - // chkArraysEqualFloat64(t, res, []float64{-2.5, -1.5, -1, -1}) - // }) - // A.Barrier() - // - // // AllReduceMax - // t.Run("AllReduceMaxFloat64s", func(t *testing.T) { - // x := make([]float64, 4) - // setSliceFloat64(x, int(A.Rank()), 3.5) - // res := make([]float64, len(x)) - // A.AllReduceMaxFloat64s(res, x) - // chkArraysEqualFloat64(t, res, []float64{4.5, 5.5, 6.5, 7.5}) - // }) - // A.Barrier() - // - // Send & Recv + t.Run("SendFloat64s/RecvFloat64s", func(t *testing.T) { if A.Rank() == 0 { s := []float64{123, 123, 123, 123} for k := 1; k <= 3; k++ { - A.SendFloat64s(s, k, 1) + A.Send(s, k, 1) } } else { - y, s := A.RecvFloat64s(0, 1) - chkStatus(s, 0, 1) - chkArraysEqualFloat64(y, []float64{123, 123, 123, 123}) + y, s := A.Recv[float64](0, 1) + if !chkStatus(s, 0, 1) { + t.Errorf("unexpected status: source %d tag %d", s.Source(), s.Tag()) + } + if !slicesEqual(y, []float64{123, 123, 123, 123}) { + t.Errorf("got %v, want %v", y, []float64{123, 123, 123, 123}) + } } }) A.Barrier() @@ -1032,203 +400,146 @@ func TestMPI(t *testing.T) { if A.Rank() == 0 { s := []int64{123, 123, 123, 123} for k := 1; k <= 3; k++ { - A.SendInt64s(s, k, 2) + A.Send(s, k, 2) } } else { - y, s := A.RecvInt64s(0, 2) - chkArraysEqualInt64(y, []int64{123, 123, 123, 123}) - chkStatus(s, 0, 2) + y, s := A.Recv[int64](0, 2) + if !slicesEqual(y, []int64{123, 123, 123, 123}) { + t.Errorf("got %v, want %v", y, []int64{123, 123, 123, 123}) + } + if !chkStatus(s, 0, 2) { + t.Errorf("unexpected status: source %d tag %d", s.Source(), s.Tag()) + } } }) - A.Barrier() - // SendOneI/RecvOneI + t.Run("SendInt64/RecvInt64", func(t *testing.T) { if A.Rank() == 0 { for k := 1; k <= 3; k++ { - A.SendInt64(int64(k*111), k, 3) + A.SendOne(int64(k*111), k, 3) } } else { - res, s := A.RecvInt64(0, 3) + res, s := A.RecvOne[int64](0, 3) exp := int64(111 * A.Rank()) if res != exp { - t.Errorf("received %d, expected %d", res, exp) + t.Errorf("got %d, want %d", res, exp) + } + if !chkStatus(s, 0, 3) { + t.Errorf("unexpected status: source %d tag %d", s.Source(), s.Tag()) } - chkStatus(s, 0, 3) } }) - A.Barrier() - // SendB / RecvB + t.Run("SendBytes/RecvBytes", func(t *testing.T) { if A.Rank() == 0 { for k := 1; k <= 3; k++ { - s := fmt.Sprintf("Hello Rank %d!", k) - A.SendBytes([]byte(s), k, 4) + A.Send(fmt.Appendf(nil, "Hello Rank %d!", k), k, 4) } } else { res := make([]byte, 13) exp := fmt.Sprintf("Hello Rank %d!", A.Rank()) - s := A.RecvPreallocBytes(res, 0, 4) + s := A.RecvPrealloc(res, 0, 4) if string(res) != exp { - t.Errorf("received %s, expected %s", res, exp) + t.Errorf("got %s, want %s", res, exp) } - chkStatus(s, 0, 4) - } - }) - - A.Barrier() - - // SendOneString / RecvOneString - t.Run("SendString/RecvString", func(t *testing.T) { - if A.Rank() == 0 { - for k := 1; k <= 3; k++ { - str := fmt.Sprintf("Hello Rank %d!", k) - A.SendString(str, k, 5) - } - } else { - res, s := A.RecvString(0, 5) - exp := fmt.Sprintf("Hello Rank %d!", A.Rank()) - if res != exp { - t.Errorf("received %s, expected %s", res, exp) + if !chkStatus(s, 0, 4) { + t.Errorf("unexpected status: source %d tag %d", s.Source(), s.Tag()) } - chkStatus(s, 0, 5) } }) - A.Barrier() - // Probe t.Run("Probe", func(t *testing.T) { if A.Rank() == 3 { vals := []int64{1, 4, 9} - for k := 0; k < 3; k++ { - A.SendInt64s(vals, k, 6) + for k := range 3 { + A.Send(vals, k, 6) } } else { s := A.Probe(3, 6) - src := s.GetSource() - if src != 3 { - t.Errorf("GetSource: received %d, expected 3", src) + if src := s.Source(); src != 3 { + t.Errorf("Source: got %d, want 3", src) } - n := s.GetCount(Long) - if n != 3 { - t.Errorf("GetCount: received %d, expected 3", n) + if n := s.Count[int64](); n != 3 { + t.Errorf("Count: got %d, want 3", n) } } }) - // Iprobe t.Run("Iprobe", func(t *testing.T) { if A.Rank() == 1 { if b, _ := A.Iprobe(3, 0); b { - t.Errorf("Status: received %v, expected false", b) + t.Errorf("Iprobe: got true, want false") } } if A.Rank() == 3 { vals := []int64{1, 4, 9} - for k := 0; k < 3; k++ { - A.SendInt64s(vals, k, 6) + for k := range 3 { + A.Send(vals, k, 6) } } else { - b, _ := A.Iprobe(3, AnyTag) - if !b { - t.Errorf("Status: received %v, expected true", b) + if b, _ := A.Iprobe(3, AnyTag); !b { + t.Errorf("Iprobe(3, AnyTag): got false, want true") } - b, _ = A.Iprobe(AnySource, 6) - if !b { - t.Errorf("Status: received %v, expected true", b) + if b, _ := A.Iprobe(AnySource, 6); !b { + t.Errorf("Iprobe(AnySource, 6): got false, want true") } - b, _ = A.Iprobe(AnySource, AnyTag) - if !b { - t.Errorf("Status: received %v, expected true", b) + if b, _ := A.Iprobe(AnySource, AnyTag); !b { + t.Errorf("Iprobe(AnySource, AnyTag): got false, want true") } b, s := A.Iprobe(3, 6) if !b { - t.Errorf("Status: received %v, expected true", b) + t.Errorf("Iprobe(3, 6): got false, want true") } - src := s.GetSource() - if src != 3 { - t.Errorf("GetSource: received %d, expected 3", src) + if src := s.Source(); src != 3 { + t.Errorf("Source: got %d, want 3", src) } - n := s.GetCount(Long) - if n != 3 { - t.Errorf("GetCount: received %d, expected 3", n) + if n := s.Count[int64](); n != 3 { + t.Errorf("Count: got %d, want 3", n) } } }) -} + A.Barrier() + + t.Run("WorldRank", func(t *testing.T) { + if m.WorldRank() != int(A.Rank()) { + t.Errorf("got %d, want %d", m.WorldRank(), A.Rank()) + } + }) + A.Barrier() + + t.Run("WorldSize", func(t *testing.T) { + if m.WorldSize() != 4 { + t.Errorf("got %d, want 4", m.WorldSize()) + } + }) + A.Barrier() -// -// } else { -// -// // BcastFromRootC -// x := make([]complex128, 8) -// if B.Rank() == 0 { -// for i := 0; i < len(x); i++ { -// x[i] = complex(float64(1+i), float64(1+i)/10.0) -// } -// } -// B.BcastFromRootC(x) -// chk.ArrayC(tst, "B: x (complex)", 1e-17, x, []complex128{1 + 0.1i, 2 + 0.2i, 3 + 0.3i, 4 + 0.4i, 5 + 0.5i, 6 + 0.6i, 7 + 0.7i, 8 + 0.8i}) -// -// // ReduceSum -// setSliceC(x, int(B.Rank()), int(B.Size())) -// res := make([]complex128, len(x)) -// B.ReduceSumC(res, x) -// if B.Rank() == 0 { -// chk.ArrayC(tst, "B root: res", 1e-17, res, []complex128{1 + 0.1i, 1 + 0.1i, 2 + 0.2i, 2 + 0.2i, 3 + 0.3i, 3 + 0.3i, 4 + 0.4i, 4 + 0.4i}) -// } else { -// chk.ArrayC(tst, "B others: res", 1e-17, res, nil) -// } -// -// // AllReduceSumC -// setSliceC(x, int(B.Rank()), int(B.Size())) -// for i := 0; i < len(x); i++ { -// res[i] = 0 -// } -// B.AllReduceSumC(res, x) -// chk.ArrayC(tst, "B all: res", 1e-17, res, []complex128{1 + 0.1i, 1 + 0.1i, 2 + 0.2i, 2 + 0.2i, 3 + 0.3i, 3 + 0.3i, 4 + 0.4i, 4 + 0.4i}) -// -// // AllReduceMinI -// z := make([]int, 8) -// zres := make([]int, 8) -// setSliceI(z, int(B.Rank()), int(B.Size())) -// B.AllReduceMinI(zres, z) -// chk.Ints(tst, "A all (min int): res", zres, []int{-1, -1, -1, -1, -1, -1, -1, -1}) -// -// // AllReduceMaxI -// setSliceI(z, int(B.Rank()), int(B.Size())) -// for i := 0; i < len(z); i++ { -// zres[i] = 0 -// } -// B.AllReduceMaxI(zres, z) -// chk.Ints(tst, "A all (max int): res", zres, []int{1, 1, 2, 2, 3, 3, 4, 4}) -// -// // SendC & RecvC -// if B.Rank() == 0 { -// s := []complex128{123 + 1i, 123 + 2i, 123 + 3i, 123 + 4i} -// for k := 1; k <= 3; k++ { -// B.SendC(s, k) -// } -// } else { -// y := make([]complex128, 4) -// B.RecvC(y, 0) -// chk.ArrayC(tst, "B recv", 1e-17, y, []complex128{123 + 1i, 123 + 2i, 123 + 3i, 123 + 4i}) -// } -// -// // SendOne & RecvOne -// if B.Rank() == 0 { -// for k := 1; k <= 3; k++ { -// B.SendOne(-123, k) -// } -// } else { -// res := B.RecvOne(0) -// chk.Float64(tst, "B RecvOne", 1e-17, res, -123) -// } -// } -// -// // wait for all -// world := mpi.NewCommunicator(nil) -// world.Barrier() -// } + t.Run("WorldTime", func(t *testing.T) { + t1 := m.WorldTime() + t2 := m.WorldTime() + if t2 < t1 { + t.Errorf("WorldTime not monotonic: %f < %f", t2, t1) + } + }) + A.Barrier() + + t.Run("Mrecv", func(t *testing.T) { + if A.Rank() == 0 { + s := []float64{1, 2, 3, 4} + for k := 1; k <= 3; k++ { + A.Send(s, k, 7) + } + } else { + y, s := A.Mrecv[float64](0, 7) + if !slicesEqual(y, []float64{1, 2, 3, 4}) { + t.Errorf("got %v, want %v", y, []float64{1, 2, 3, 4}) + } + if !chkStatus(s, 0, 7) { + t.Errorf("unexpected status: source %d tag %d", s.Source(), s.Tag()) + } + } + }) +} diff --git a/op_string.go b/op_string.go deleted file mode 100644 index 61ed860..0000000 --- a/op_string.go +++ /dev/null @@ -1,32 +0,0 @@ -// Code generated by "stringer -type=Op"; DO NOT EDIT. - -package mpi - -import "strconv" - -func _() { - // An "invalid array index" compiler error signifies that the constant values have changed. - // Re-run the stringer command to generate them again. - var x [1]struct{} - _ = x[OpSum-0] - _ = x[OpMin-1] - _ = x[OpMax-2] - _ = x[OpProd-3] - _ = x[OpLand-4] - _ = x[OpLor-5] - _ = x[OpLxor-6] - _ = x[OpBand-7] - _ = x[OpBor-8] - _ = x[OpBxor-9] -} - -const _Op_name = "OpSumOpMinOpMaxOpProdOpLandOpLorOpLxorOpBandOpBorOpBxor" - -var _Op_index = [...]uint8{0, 5, 10, 15, 21, 27, 32, 38, 44, 49, 55} - -func (i Op) String() string { - if i >= Op(len(_Op_index)-1) { - return "Op(" + strconv.FormatInt(int64(i), 10) + ")" - } - return _Op_name[_Op_index[i]:_Op_index[i+1]] -} diff --git a/ops.go b/ops.go new file mode 100644 index 0000000..30b49ae --- /dev/null +++ b/ops.go @@ -0,0 +1,67 @@ +//go:build !windows + +package mpi + +/* +#include "mpi.h" +*/ +import "C" + +import "strconv" + +// Op identifies an MPI reduction operation. +type Op uint8 + +// Op constants identify the supported MPI reduction operations. +const ( + OpSum Op = iota // MPI_SUM + OpMin // MPI_MIN + OpMax // MPI_MAX + OpProd // MPI_PROD + OpLand // MPI_LAND (logical and) + OpLor // MPI_LOR (logical or) + OpLxor // MPI_LXOR (logical xor) + OpBand // MPI_BAND (bitwise and) + OpBor // MPI_BOR (bitwise or) + OpBxor // MPI_BXOR (bitwise xor) +) + +func (o Op) String() string { + switch o { + case OpSum: + return "OpSum" + case OpMin: + return "OpMin" + case OpMax: + return "OpMax" + case OpProd: + return "OpProd" + case OpLand: + return "OpLand" + case OpLor: + return "OpLor" + case OpLxor: + return "OpLxor" + case OpBand: + return "OpBand" + case OpBor: + return "OpBor" + case OpBxor: + return "OpBxor" + default: + return "Op(" + strconv.Itoa(int(o)) + ")" + } +} + +var ops = [...]C.MPI_Op{ + C.MPI_SUM, + C.MPI_MIN, + C.MPI_MAX, + C.MPI_PROD, + C.MPI_LAND, + C.MPI_LOR, + C.MPI_LXOR, + C.MPI_BAND, + C.MPI_BOR, + C.MPI_BXOR, +} diff --git a/types.go b/types.go new file mode 100644 index 0000000..d43369c --- /dev/null +++ b/types.go @@ -0,0 +1,126 @@ +//go:build !windows + +package mpi + +/* +#include "mpi.h" + +#define DOUBLE_COMPLEX double complex +*/ +import "C" + + +type dataType uint8 + +const ( + dtInt8 dataType = iota // MPI_INT8_T → int8 + dtByte // MPI_BYTE → byte (uint8) + dtInt16 // MPI_INT16_T → int16 + dtUint16 // MPI_UINT16_T → uint16 + dtInt // MPI_INT32_T → int32 + dtUint // MPI_UINT32_T → uint32 + dtLong // MPI_INT64_T → int64 + dtUlong // MPI_UINT64_T → uint64 + dtFloat // MPI_FLOAT → float32 + dtDouble // MPI_DOUBLE → float64 + dtFloatComplex // MPI_C_FLOAT_COMPLEX → complex64 + dtComplex // MPI_DOUBLE_COMPLEX → complex128 +) + +// mpiDataTypes maps each dataType constant to its corresponding C.MPI_Datatype. +// +// MPI type handles are opaque C global variables, not compile-time constants, +// so they cannot be embedded directly in Go const expressions. Any function +// that references them (e.g. via a type switch returning C.MPI_BYTE) incurs +// a cgo call cost that drives the Go inliner budget far above its threshold +// of 80 nodes, preventing inlining entirely. +// +// By loading the handles once into this array at init time, all subsequent +// lookups are plain indexed memory reads with no cgo overhead, and the +// dispatch functions (goDataType, dataTypeOf) remain inlinable. +var mpiDataTypes = [...]C.MPI_Datatype{ + C.MPI_INT8_T, + C.MPI_BYTE, + C.MPI_INT16_T, + C.MPI_UINT16_T, + C.MPI_INT32_T, + C.MPI_UINT32_T, + C.MPI_INT64_T, + C.MPI_UINT64_T, + C.MPI_FLOAT, + C.MPI_DOUBLE, + C.MPI_C_FLOAT_COMPLEX, + C.MPI_DOUBLE_COMPLEX, +} + +type goTypes interface { + int8 | byte | int16 | uint16 | int32 | uint32 | int64 | uint64 | float32 | float64 | complex64 | complex128 +} + +// goDataType returns the dataType index for T. It contains no cgo calls and +// is inlinable (verified via -gcflags="-m=2"; inliner cost ~36, budget 80). +// Returning a plain integer here rather than a C.MPI_Datatype is what keeps +// the cost low: see the mpiDataTypes comment for the full rationale. +func goDataType[T goTypes]() dataType { + var zero T + switch any(zero).(type) { + case int8: + return dtInt8 + case byte: + return dtByte + case int16: + return dtInt16 + case uint16: + return dtUint16 + case int32: + return dtInt + case uint32: + return dtUint + case int64: + return dtLong + case uint64: + return dtUlong + case float32: + return dtFloat + case float64: + return dtDouble + case complex64: + return dtFloatComplex + case complex128: + return dtComplex + default: + panic("unreachable") + } +} + +// dataTypeOf returns the C.MPI_Datatype for T via a single array lookup. +// Inlinable at cost ~45. +func dataTypeOf[T goTypes]() C.MPI_Datatype { + return mpiDataTypes[goDataType[T]()] +} + +// isValidForOp reports whether op is valid for the given dataType. +func isValidForOp(dt dataType, o Op) bool { + switch dt { + case dtFloat, dtDouble: + return o != OpLand && o != OpLor && o != OpLxor && + o != OpBand && o != OpBor && o != OpBxor + case dtFloatComplex, dtComplex: + return o == OpSum || o == OpProd + default: + return true + } +} + +// isValidDataTypeForOp reports whether op is valid for the Go type T. +func isValidDataTypeForOp[T goTypes](o Op) bool { + return isValidForOp(goDataType[T](), o) +} + +// getDataTypeAndValidate returns the C.MPI_Datatype for T and whether op is +// valid for that type. It calls goDataType once, avoiding the double type +// dispatch that separate dataTypeOf + isValidDataTypeForOp calls would incur. +func getDataTypeAndValidate[T goTypes](o Op) (C.MPI_Datatype, bool) { + dt := goDataType[T]() + return mpiDataTypes[dt], isValidForOp(dt, o) +}