+++ /dev/null
-# This file is part of NIT (http://www.nitlanguage.org).
-#
-# Copyright 2014 Alexis Laferrière <alexis.laf@xymus.net>
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Implementation of the Message Passing Interface protocol by wrapping OpenMPI
-#
-# OpenMPI is used only at linking and for it's `mpi.h`. Other implementations
-# could be used without much modification.
-#
-# Supports transfer of any valid `Serializable` instances as well as basic
-# C arrays defined in module `c`. Using C arrays is encouraged when performance
-# is critical.
-#
-# Since this module is a thin wrapper around OpenMPI, in case of missing
-# documentation, you can refer to https://www.open-mpi.org/doc/v1.8/.
-module mpi is
- cflags exec("mpicc", "-showme:compile")
- ldflags exec("mpicc", "-showme:link")
-end
-
-import c
-intrude import core::text::flat
-import serialization
-private import json::serialization
-
-in "C Header" `{
- #include <mpi.h>
-`}
-
-# Handle to most MPI services
-class MPI
- # Initialize the MPI execution environment
- init do native_init
-
- private fun native_init `{ MPI_Init(NULL, NULL); `}
-
- # Terminates the MPI execution environment
- fun finalize `{ MPI_Finalize(); `}
-
- # Name of this processor, usually the hostname
- fun processor_name: String import NativeString.to_s_with_length `{
- char *name = malloc(MPI_MAX_PROCESSOR_NAME);
- int size;
- MPI_Get_processor_name(name, &size);
- return NativeString_to_s_with_length(name, size);
- `}
-
- # Send the content of a buffer
- fun send_from(buffer: Sendable, at, count: Int, dest: Rank, tag: Tag, comm: Comm)
- do
- buffer.send(self, at, count, dest, tag, comm)
- end
-
- # Send the full content of a buffer
- fun send_all(buffer: Sendable, dest: Rank, tag: Tag, comm: Comm)
- do
- buffer.send_all(self, dest, tag, comm)
- end
-
- # Efficiently receive data in an existing buffer
- fun recv_into(buffer: Receptacle, at, count: Int, source: Rank, tag: Tag, comm: Comm)
- do
- buffer.recv(self, at, count, source, tag, comm)
- end
-
- # Efficiently receive data and fill an existing buffer
- fun recv_fill(buffer: Receptacle, source: Rank, tag: Tag, comm: Comm)
- do
- buffer.recv_fill(self, source, tag, comm)
- end
-
- # Send a complex `Serializable` object
- fun send(data: nullable Serializable, dest: Rank, tag: Tag, comm: Comm)
- do
- # Serialize data
- var stream = new StringWriter
- var serializer = new JsonSerializer(stream)
- serializer.serialize(data)
-
- # Send message
- var str = stream.to_s
- send_from(str, 0, str.length, dest, tag, comm)
- end
-
- # Receive a complex object
- fun recv(source: Rank, tag: Tag, comm: Comm): nullable Object
- do
- var status = new Status
-
- # Block until a message in in queue
- var err = probe(source, tag, comm, status)
- assert err.is_success else print err
-
- # Get message length
- var count = status.count(new DataType.char)
- assert not count.is_undefined
-
- # Receive message into buffer
- var buffer = new FlatBuffer.with_capacity(count)
- recv_into(buffer, 0, count, status.source, status.tag, comm)
-
- # Free our status
- status.free
-
- # Deserialize message
- var deserializer = new JsonDeserializer(buffer)
- var deserialized = deserializer.deserialize
-
- if deserialized == null then print "|{buffer}|{buffer.chars.join("-")}| {buffer.length}"
-
- return deserialized
- end
-
- # Send an empty buffer, only for the `tag`
- fun send_empty(dest: Rank, tag: Tag, comm: Comm): SuccessOrError
- `{
- return MPI_Send(NULL, 0, MPI_CHAR, dest, tag, comm);
- `}
-
- # Receive an empty buffer, only for the `tag`
- fun recv_empty(dest: Rank, tag: Tag, comm: Comm): SuccessOrError
- `{
- return MPI_Recv(NULL, 0, MPI_CHAR, dest, tag, comm, MPI_STATUS_IGNORE);
- `}
-
- # Send a `NativeCArray` `buffer` with a given `count` of `data_type`
- fun native_send(buffer: NativeCArray, count: Int, data_type: DataType, dest: Rank, tag: Tag, comm: Comm): SuccessOrError
- `{
- return MPI_Send(buffer, count, data_type, dest, tag, comm);
- `}
-
- # Receive into a `NativeCArray` `buffer` with a given `count` of `data_type`
- fun native_recv(buffer: NativeCArray, count: Int, data_type: DataType, dest: Rank, tag: Tag, comm: Comm, status: Status): SuccessOrError
- `{
- return MPI_Recv(buffer, count, data_type, dest, tag, comm, status);
- `}
-
- # Probe for the next data to receive, store the result in `status`
- #
- # Note: If you encounter an error where the next receive does not correspond
- # to the last `probe`, call this method twice to ensure a correct result.
- fun probe(source: Rank, tag: Tag, comm: Comm, status: Status): SuccessOrError
- `{
- return MPI_Probe(source, tag, comm, status);
- `}
-
- # Synchronize all processors
- fun barrier(comm: Comm) `{ MPI_Barrier(comm); `}
-
- # Seconds since some time in the past which does not change
- fun wtime: Float `{ return MPI_Wtime(); `}
-end
-
-# An MPI communicator
-extern class Comm `{ MPI_Comm `}
- # The _null_ communicator, targeting no processors
- new null_ `{ return MPI_COMM_NULL; `}
-
- # The _world_ communicator, targeting all processors
- new world `{ return MPI_COMM_WORLD; `}
-
- # The _self_ communicator, targeting this processor only
- new self_ `{ return MPI_COMM_SELF; `}
-
- # Number of processors in this communicator
- fun size: Int `{
- int size;
- MPI_Comm_size(self, &size);
- return size;
- `}
-
- # Rank on this processor in this communicator
- fun rank: Rank `{
- int rank;
- MPI_Comm_rank(self, &rank);
- return rank;
- `}
-end
-
-# An MPI data type
-extern class DataType `{ MPI_Datatype `}
- # Get a MPI char.
- new char `{ return MPI_CHAR; `}
-
- # Get a MPI short.
- new short `{ return MPI_SHORT; `}
-
- # Get a MPI int.
- new int `{ return MPI_INT; `}
-
- # Get a MPI long.
- new long `{ return MPI_LONG; `}
-
- # Get a MPI long long.
- new long_long `{ return MPI_LONG_LONG; `}
-
- # Get a MPI unsigned char.
- new unsigned_char `{ return MPI_UNSIGNED_CHAR; `}
-
- # Get a MPI unsigned short.
- new unsigned_short `{ return MPI_UNSIGNED_SHORT; `}
-
- # Get a MPI unsigned int.
- new unsigned `{ return MPI_UNSIGNED; `}
-
- # Get a MPI unsigned long.
- new unsigned_long `{ return MPI_UNSIGNED_LONG; `}
-
- # Get a MPI unsigned long long.
- new unsigned_long_long `{ return MPI_UNSIGNED_LONG_LONG; `}
-
- # Get a MPI float.
- new float `{ return MPI_FLOAT; `}
-
- # Get a MPI double.
- new double `{ return MPI_DOUBLE; `}
-
- # Get a MPI long double.
- new long_double `{ return MPI_LONG_DOUBLE; `}
-
- # Get a MPI byte.
- new byte `{ return MPI_BYTE; `}
-end
-
-# Status of a communication used by `MPI::probe`
-extern class Status `{ MPI_Status* `}
- # Ignore the resulting status
- new ignore `{ return MPI_STATUS_IGNORE; `}
-
- # Allocated a new `Status`, must be freed with `free`
- new `{ return malloc(sizeof(MPI_Status)); `}
-
- # Source of this communication
- fun source: Rank `{ return self->MPI_SOURCE; `}
-
- # Tag of this communication
- fun tag: Tag `{ return self->MPI_TAG; `}
-
- # Success or error on this communication
- fun error: SuccessOrError `{ return self->MPI_ERROR; `}
-
- # Count of the given `data_type` in this communication
- fun count(data_type: DataType): Int
- `{
- int count;
- MPI_Get_count(self, data_type, &count);
- return count;
- `}
-end
-
-# An MPI operation
-#
-# Used with the `reduce` method.
-#
-# See <http://www.mpi-forum.org/docs/mpi-1.1/mpi-11-html/node78.html>
-extern class Op `{ MPI_Op `}
- # Get a MPI null operation.
- new op_null `{ return MPI_OP_NULL; `}
-
- # Get a MPI maximum operation.
- new max `{ return MPI_MAX; `}
-
- # Get a MPI minimum operation.
- new min `{ return MPI_MIN; `}
-
- # Get a MPI sum operation.
- new sum `{ return MPI_SUM; `}
-
- # Get a MPI product operation.
- new prod `{ return MPI_PROD; `}
-
- # Get a MPI logical and operation.
- new land `{ return MPI_LAND; `}
-
- # Get a MPI bit-wise and operation.
- new band `{ return MPI_BAND; `}
-
- # Get a MPI logical or operation.
- new lor `{ return MPI_LOR; `}
-
- # Get a MPI bit-wise or operation.
- new bor `{ return MPI_BOR; `}
-
- # Get a MPI logical xor operation.
- new lxor `{ return MPI_LXOR; `}
-
- # Get a MPI bit-wise xor operation.
- new bxor `{ return MPI_BXOR; `}
-
- # Get a MPI minloc operation.
- #
- # Used to compute a global minimum and also an index attached
- # to the minimum value.
- #
- # See <http://www.mpi-forum.org/docs/mpi-1.1/mpi-11-html/node79.html#Node79>
- new minloc `{ return MPI_MINLOC; `}
-
- # Get a MPI maxloc operation.
- #
- # Used to compute a global maximum and also an index attached
- # to the maximum value.
- #
- # See <http://www.mpi-forum.org/docs/mpi-1.1/mpi-11-html/node79.html#Node79>
- new maxloc `{ return MPI_MAXLOC; `}
-
- # Get a MPI replace operation.
- new replace `{ return MPI_REPLACE; `}
-end
-
-# An MPI return code to report success or errors
-extern class SuccessOrError `{ int `}
- # Is this a success?
- fun is_success: Bool `{ return self == MPI_SUCCESS; `}
-
- # Is this an error?
- fun is_error: Bool do return not is_success
-
- # TODO add is_... for each variant
-
- # The class of this error
- fun error_class: ErrorClass
- `{
- int class;
- MPI_Error_class(self, &class);
- return class;
- `}
-
- redef fun to_s do return native_to_s.to_s
- private fun native_to_s: NativeString `{
- char *err = malloc(MPI_MAX_ERROR_STRING);
- MPI_Error_string(self, err, NULL);
- return err;
- `}
-end
-
-# An MPI error class
-extern class ErrorClass `{ int `}
- redef fun to_s do return native_to_s.to_s
- private fun native_to_s: NativeString `{
- char *err = malloc(MPI_MAX_ERROR_STRING);
- MPI_Error_string(self, err, NULL);
- return err;
- `}
-end
-
-# An MPI rank within a communcator
-extern class Rank `{ int `}
- # Special rank accepting any processor
- new any `{ return MPI_ANY_SOURCE; `}
-
- # This Rank as an `Int`
- fun to_i: Int `{ return self; `}
- redef fun to_s do return to_i.to_s
-end
-
-# An MPI tag, can be defined using `Int::tag`
-extern class Tag `{ int `}
- # Special tag accepting any tag
- new any `{ return MPI_ANY_TAG; `}
-
- # This tag as an `Int`
- fun to_i: Int `{ return self; `}
- redef fun to_s do return to_i.to_s
-end
-
-redef universal Int
- # `self`th MPI rank
- fun rank: Rank `{ return self; `}
-
- # Tag identified by `self`
- fun tag: Tag `{ return self; `}
-
- # Is this value undefined according to MPI? (may be returned by `Status::count`)
- fun is_undefined: Bool `{ return self == MPI_UNDEFINED; `}
-end
-
-# Something sendable directly and efficiently over MPI
-#
-# Subclasses of `Sendable` should use the native MPI send function, without
-# using Nit serialization.
-interface Sendable
- # Type specific send over MPI
- protected fun send(mpi: MPI, at, count: Int, dest: Rank, tag: Tag, comm: Comm) is abstract
-
- # Type specific send full buffer over MPI
- protected fun send_all(mpi: MPI, dest: Rank, tag: Tag, comm: Comm) is abstract
-end
-
-
-# Something which can receive data directly and efficiently from MPI
-#
-# Subclasses of `Receptacle` should use the native MPI recveive function,
-# without using Nit serialization.
-interface Receptacle
- # Type specific receive from MPI
- protected fun recv(mpi: MPI, at, count: Int, source: Rank, tag: Tag, comm: Comm) is abstract
-
- # Type specific receive and fill buffer from MPI
- protected fun recv_fill(mpi: MPI, source: Rank, tag: Tag, comm: Comm) is abstract
-end
-
-redef class CArray[E]
- super Sendable
- super Receptacle
-end
-
-redef class Text
- super Sendable
-
- redef fun send(mpi, at, count, dest, tag, comm)
- do
- var str
- if at != 0 or count != length then
- str = substring(at, count)
- else str = self
-
- mpi.native_send(str.to_cstring, count, new DataType.char,
- dest, tag, new Comm.world)
- end
-
- redef fun send_all(mpi, dest, tag, comm) do send(mpi, 0, length, dest, tag, comm)
-end
-
-redef class FlatBuffer
- super Receptacle
-
- redef fun recv(mpi, at, count, source, tag, comm)
- do
- var min_capacity = at + count
- if capacity < min_capacity then enlarge min_capacity
-
- var array
- if at != 0 then
- array = items + at
- else array = items
-
- mpi.native_recv(array, count, new DataType.char,
- source, tag, new Comm.world, new Status.ignore)
-
- length = capacity
- is_dirty = true
- end
-
- redef fun recv_fill(mpi, dest, tag, comm) do recv(mpi, 0, capacity, dest, tag, comm)
-end
-
-redef class CIntArray
- redef fun send(mpi, at, count, dest, tag, comm)
- do
- var array
- if at != 0 then
- array = native_array + at
- else array = native_array
-
- mpi.native_send(array, count, new DataType.int,
- dest, tag, new Comm.world)
- end
-
- redef fun send_all(mpi, dest, tag, comm) do send(mpi, 0, length, dest, tag, comm)
-
- redef fun recv(mpi, at, count, source, tag, comm)
- do
- var array
- if at != 0 then
- array = native_array + at
- else array = native_array
-
- mpi.native_recv(array, count, new DataType.int,
- source, tag, new Comm.world, new Status.ignore)
- end
-
- redef fun recv_fill(mpi, dest, tag, comm) do recv(mpi, 0, length, dest, tag, comm)
-end
-
-# Shortcut to the world communicator (same as `new Comm.world`)
-fun comm_world: Comm do return once new Comm.world
nitlanguage / nit.git / blobdiff
