text:
define([
	'dojo/request',
	'dojo/when',
	'dojo/_base/lang',
	'dojo/json',
	'dojo/io-query',
	'dojo/_base/declare',
	'./Request' /*=====, './Store' =====*/
], function (request, when, lang, JSON, ioQuery, declare, Request /*=====, Store =====*/) {

	/*=====
	var __HeaderOptions = {
			// headers: Object?
			//		Additional headers to send along with the request.
		},
		__PutDirectives = declare(Store.PutDirectives, __HeaderOptions),
	=====*/

	return declare(Request, {

		// stringify: Function
		//		This function performs the serialization of the data for requests to the server. This
		//		defaults to JSON, but other formats can be serialized by providing an alternate
		//		stringify function. If you do want to use an alternate format, you will probably
		//		want to use an alternate parse function for the parsing of data as well.
		stringify: JSON.stringify,

		get: function (id, options) {
			// summary:
			//		Retrieves an object by its identity. This will trigger a GET request to the server using
			//		the url `this.target + id`.
			// id: Number
			//		The identity to use to lookup the object
			// options: Object?
			//		HTTP headers. For consistency with other methods, if a `headers` key exists on this
			//		object, it will be used to provide HTTP headers instead.
			// returns: Object
			//		The object in the store that matches the given id.
			options = options || {};
			var headers = lang.mixin({ Accept: this.accepts }, this.headers, options.headers || options);
			var store = this;
			return request(this.target + id, {
				headers: headers
			}).then(function (response) {
				return store._restore(store.parse(response), true);
			});
		},

		autoEmitEvents: false, // this is handled by the methods themselves

		put: function (object, options) {
			// summary:
			//		Stores an object. This will trigger a PUT request to the server
			//		if the object has an id, otherwise it will trigger a POST request.
			// object: Object
			//		The object to store.
			// options: __PutDirectives?
			//		Additional metadata for storing the data.  Includes an 'id'
			//		property if a specific id is to be used.
			// returns: dojo/_base/Deferred
			options = options || {};
			var id = ('id' in options) ? options.id : this.getIdentity(object);
			var hasId = typeof id !== 'undefined';
			var store = this;

			var positionHeaders = 'beforeId' in options
				? (options.beforeId === null
					? { 'Put-Default-Position': 'end' }
					: { 'Put-Before': options.beforeId })
				: (!hasId || options.overwrite === false
					? { 'Put-Default-Position': (this.defaultNewToStart ? 'start' : 'end') }
					: null);

			var initialResponse = request(hasId ? this.target + id : this.target, {
				method: hasId && !options.incremental ? 'PUT' : 'POST',
				data: this.stringify(object),
				headers: lang.mixin({
					'Content-Type': 'application/json',
					Accept: this.accepts,
					'If-Match': options.overwrite === true ? '*' : null,
					'If-None-Match': options.overwrite === false ? '*' : null
				}, positionHeaders, this.headers, options.headers)
			});
			return initialResponse.then(function (response) {
				var event = {};

				if ('beforeId' in options) {
					event.beforeId = options.beforeId;
				}

				var result = event.target = store._restore(store.parse(response), true) || object;

				when(initialResponse.response, function (httpResponse) {
					store.emit(httpResponse.status === 201 ? 'add' : 'update', event);
				});

				return result;
			});
		},

		add: function (object, options) {
			// summary:
			//		Adds an object. This will trigger a PUT request to the server
			//		if the object has an id, otherwise it will trigger a POST request.
			// object: Object
			//		The object to store.
			// options: __PutDirectives?
			//		Additional metadata for storing the data.  Includes an 'id'
			//		property if a specific id is to be used.
			options = options || {};
			options.overwrite = false;
			return this.put(object, options);
		},

		remove: function (id, options) {
			// summary:
			//		Deletes an object by its identity. This will trigger a DELETE request to the server.
			// id: Number
			//		The identity to use to delete the object
			// options: __HeaderOptions?
			//		HTTP headers.
			options = options || {};
			var store = this;
			return request(this.target + id, {
				method: 'DELETE',
				headers: lang.mixin({}, this.headers, options.headers)
			}).then(function (response) {
				var target = response && store.parse(response);
				store.emit('remove', {id: id, target: target});
				return response ? target : true;
			});
		}
	});

});

redpajama_set_name: RedPajamaGithub


text:
<?php get_header(); //get header ?>

                    <?php
                    $args = array(
                        'post_type' => 'services',
                        'category_name' => 'sliders',
                        'posts_per_page' => 3
                    );
                    $sliders = new WP_Query($args); ?>

                    <?php if ($sliders->have_posts()) {
                        while ($sliders->have_posts()) {
                            $sliders->the_post();
                            ?>
                            <li data-target="#carousel-example-generic"
                                data-slide-to="<?php echo $i; ?>" class="<?= ($sliders->current_post == 0) ? 'active' : '' ?>">

                                <?php
                        }
                    } else {
                        ?>
<?php _e('Sorry, no posts matched your criteria.'); ?>
<?php
                    } ?>

                    <?php if ($sliders->have_posts()) {
                        while ($sliders->have_posts()) {
                            $sliders->the_post();
                            ?>

                            <div class="item <?= ($sliders->current_post == 0) ? 'active' : '' ?>">
                                <img src="<?php echo get_the_post_thumbnail_url(get_the_ID(), 'full'); ?>"
                                     alt="<?php echo get_post_meta(get_the_ID(), 'alt_img', true); ?>">

                                            <?php the_title(); ?>

                                            <?php the_content(); ?>

                            <?php
                            wp_reset_postdata();
                        }
                    } else {
                        ?>
<?php _e('Sorry, no posts matched your criteria.'); ?>
<?php
                    } ?>

                <?php
                $args = array(
                    'post_type' => 'services',
                    'category_name' => 'services',
                    'posts_per_page' => 3
                );
                $services = new WP_Query($args); ?>

                <?php if ($services->have_posts()) {
                    while ($services->have_posts()) {
                        $services->the_post();
                        ?>

                                <i class="<?php echo get_post_meta(get_the_ID(), 'font_icon', true); ?> <?php echo get_post_meta(get_the_ID(), 'color_icon', true); ?>">
                                <?php the_title(); ?>

<?php the_content(); ?>

                        <?php
                        wp_reset_postdata();
                    }
                } else {
                    ?>
<?php _e('Sorry, no posts matched your criteria.'); ?>
<?php
                } ?>

Recent Works

Lorem ipsum dolor sit amet, dolore eiusmod quis tempor incididunt ut et dolore Ut veniam unde
                        voluptatem. Sed unde omnis iste natus error sit voluptatem.

                        <?php
                        $args = array(
                            'post_type' => 'services',
                            'category_name' => 'recent',
                            'posts_per_page' => 3
                        );
                        $recent = new WP_Query($args); ?>

                        <?php if ($recent->have_posts()) {
                            while ($recent->have_posts()) {
                                $recent->the_post();
                                ?>

                                        <img src="<?php echo get_the_post_thumbnail_url(get_the_ID(), 'full'); ?>"
                                             alt="<?php echo get_post_meta(get_the_ID(), 'alt_ind', true); ?>"
                                             class="img-responsive">

                                        <a href="<?php echo get_the_post_thumbnail_url(get_the_ID(), 'full'); ?>"
                                           class="fancybox-button"
                                           title="<?php echo get_post_meta(get_the_ID(), 'title_img', true); ?>"
                                           data-rel="fancybox-button">

                                        <?php the_title(); ?>
                                        <?php echo get_post_meta(get_the_ID(), 'corp', true); ?>

                                <?php
                                wp_reset_postdata();
                            }
                        } else {
                            ?>
<?php _e('Sorry, no posts matched your criteria.'); ?>
<?php
                        } ?>

                            <?php
                            $args = array(
                                'post_type' => 'services',
                                'category_name' => 'our-clients'
                            );
                            $clients = new WP_Query($args); ?>

                            <?php if ($clients->have_posts()) {
                            while ($clients->have_posts()){
                            $clients->the_post();
                            ?>

<?php the_title(); ?>

<?php the_content(); ?>

                        <?php
                        wp_reset_postdata();
                        }
                        } else {
                            ?>
<?php _e('Sorry, no clients matched your criteria.'); ?>
<?php
                        } ?>

                            <?php
                            $args = array(
                                'post_type' => 'services',
                                'category_name' => 'our-clients_img',
                                'posts_per_page' => 6
                            );
                            $ourclientsimg = new WP_Query($args); ?>

                            <?php if ($ourclientsimg->have_posts()) {
                                while ($ourclientsimg->have_posts()) {
                                    $ourclientsimg->the_post();
                                    ?>

                                            <img src="<?php echo get_the_post_thumbnail_url(get_the_ID(), 'full'); ?>"
                                                 class="img-responsive" alt="">
                                            <img src="<?php echo get_the_post_thumbnail_url(get_the_ID(), 'full'); ?>"
                                                 class="color-img img-responsive" alt="">

                                    <?php
                                    wp_reset_postdata();
                                }
                            } else {
                                ?>
<?php _e('Sorry, no our-clients_img matched your criteria.'); ?>
<?php
                            } ?>

                        <?php
                        $args = array(
                            'post_type' => 'services',
                            'category_name' => 'clients',
                            'posts_per_page' => 3
                        );
                        $clients = new WP_Query($args); ?>

                        <?php if ($clients->have_posts()) {
                            while ($clients->have_posts()) {
                                $clients->the_post();
                                ?>

<?php the_content(); ?>

                                        <img class="pull-left"
                                             src="<?php echo get_template_directory_uri(); ?>/assets/pages/img/people/img1-small.jpg"
                                             alt="">

                                            <?php the_title(); ?>
                                            <?php the_title(); ?>

                                <?php
                                wp_reset_postdata();
                            }
                        } else {
                            ?>
<?php _e('Sorry, no posts matched your criteria.'); ?>
<?php
                        } ?>

<?php get_footer(); ?>


text:
using NHibernate.Mapping.ByCode;
using NHibernate.Mapping.ByCode.Conformist;

namespace RegTesting.Contracts.Domain
{
	/// <summary>
	/// Classmapping for Testsystem
	/// </summary>
	public class TestsystemMap : ClassMapping<Testsystem>
	{
		/// <summary>
		/// Mapping of Testsystem
		/// </summary>
		public TestsystemMap()
		{
			Id(x => x.ID, map => map.Generator(Generators.Identity));
			Property(x => x.Description);
			Property(x => x.Name);
			Property(x => x.Url);
			Property(x => x.LastUpdated);
			Bag(p => p.Testsuites, map => map.Key(k => k.Column("TestsystemFilter")), ce => ce.OneToMany());

			Table("Testsystems");
		}
	}
}


text:
<?php
class Service_user extends CI_Model
{

    public function __construct()
    {
        $this->load->database();
        parent::__construct();
    }

    public function register_user($param)
    {
        $data[] = $param['username'];
        $data[] = $param['address'];
        $data[] = $param['phone'];
        $data[] = $param['email'];

        $this->db->trans_start();//Used for upload several datas as a PACKAGE! Start

        $sql = <<<EOF
            INSERT INTO `user` (`username`,`address`,`phone`,`email`) VALUES (?,?,?,?);
EOF;
        $res = $this->db->query($sql,$data);
        $last_user_id = $this->db->insert_id();

        $industry   = $param['input_industry'];
        $occupation = $param['input_occupation'];

        foreach ($industry as $key => $val)
        {
            $sql2 = <<<EOF
                INSERT INTO `u_i_relation` (`user_id`,`industry_id`) VALUES ('{$last_user_id}','{$val[0]}')
EOF;
            $res = $this->db->query($sql2);
        }

        foreach ($occupation as $key => $val)
        {
            $sql3 = <<<EOF
                INSERT INTO `u_o_relation` (`user_id`,`occupation_id`) VALUES ('{$last_user_id}','{$val[0]}')
EOF;
            $res = $this->db->query($sql3);
        }

        $this->db->trans_complete();//Used for upload several datas as a PACKAGE! FINISH
        return ($this->db->trans_status());

    }

    public function get_user_list()
    {
        $sql = <<<EOF
                 SELECT DISTINCT u.id,username
            FROM
                u_i_relation AS uir
                INNER JOIN user AS u
                    ON uir.user_id = u.id

                INNER JOIN u_o_relation AS uor
                    ON uor.user_id = u.id

            WHERE uir.user_id is not null AND uor.user_id is not null
                ;
EOF;

        $result = $this->db->query($sql)->result_array();
        return $result;
    }

    public function get_user_detail($id)
    {
        $this->db->trans_start();//Used for upload several datas as a PACKAGE! Start
        //
        $sql = <<<EOF
            SELECT *
            FROM
               user
            WHERE id = '{$id}'
EOF;
        $res['user'] = $this->db->query($sql)->result_array();

        $sql2 = <<<EOF
            SELECT i.industry_name
            FROM
                u_i_relation AS uir
                INNER JOIN industry AS i
                    ON uir.industry_id = i.industry_id
            WHERE uir.user_id = '{$id}'
EOF;
        $res['industry'] = $this->db->query($sql2)->result_array();

        $sql3 = <<<EOF
            SELECT o.occupation_name
            FROM
                u_o_relation AS uor
                INNER JOIN occupation AS o
                    ON uor.occupation_id = o.occupation_id
            WHERE uor.user_id = '{$id}'
EOF;
        $res['occupation'] = $this->db->query($sql3)->result_array();
        $this->db->trans_complete();
        return $res;
    }
}
?>


text:
package client

import (
	"crypto/x509"
	"encoding/base64"
	"fmt"
	"io/ioutil"
	"os"
	"path"
	"path/filepath"
	"strings"
	"time"

	"golang.org/x/net/context"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/health/grpc_health_v1"

	// Import registers the grpc GZIP encoder
	_ "google.golang.org/grpc/encoding/gzip"
	"google.golang.org/grpc/keepalive"
	"google.golang.org/grpc/metadata"

	types "github.com/gogo/protobuf/types"
	log "github.com/sirupsen/logrus"

	"github.com/pachyderm/pachyderm/v2/src/admin"
	"github.com/pachyderm/pachyderm/v2/src/auth"
	"github.com/pachyderm/pachyderm/v2/src/client/limit"
	"github.com/pachyderm/pachyderm/v2/src/debug"
	"github.com/pachyderm/pachyderm/v2/src/enterprise"
	"github.com/pachyderm/pachyderm/v2/src/identity"
	"github.com/pachyderm/pachyderm/v2/src/internal/config"
	"github.com/pachyderm/pachyderm/v2/src/internal/errors"
	"github.com/pachyderm/pachyderm/v2/src/internal/grpcutil"
	"github.com/pachyderm/pachyderm/v2/src/internal/tls"
	"github.com/pachyderm/pachyderm/v2/src/internal/tracing"
	"github.com/pachyderm/pachyderm/v2/src/license"
	"github.com/pachyderm/pachyderm/v2/src/pfs"
	"github.com/pachyderm/pachyderm/v2/src/pps"
	"github.com/pachyderm/pachyderm/v2/src/proxy"
	"github.com/pachyderm/pachyderm/v2/src/transaction"
	"github.com/pachyderm/pachyderm/v2/src/version/versionpb"
)

const (
	// MaxListItemsLog specifies the maximum number of items we log in response to a List* API
	MaxListItemsLog = 10
	// StorageSecretName is the name of the Kubernetes secret in which storage credentials are stored.
	StorageSecretName = "pachyderm-storage-secret"
	// PostgresSecretName is the name of the secret containing the postgres password
	// It must match the secret passed to pachd here: etc/helm/pachyderm/templates/pachd/deployment.yaml
	PostgresSecretName = "postgres"
	// PachctlSecretName is the name of the Kubernetes secret in which
	// pachctl credentials are stored.
	PachctlSecretName = "pachyderm-pachctl-secret"
)

// PfsAPIClient is an alias for pfs.APIClient.
type PfsAPIClient pfs.APIClient

// PpsAPIClient is an alias for pps.APIClient.
type PpsAPIClient pps.APIClient

// AuthAPIClient is an alias of auth.APIClient
type AuthAPIClient auth.APIClient

// IdentityAPIClient is an alias of identity.APIClient
type IdentityAPIClient identity.APIClient

// VersionAPIClient is an alias of versionpb.APIClient
type VersionAPIClient versionpb.APIClient

// AdminAPIClient is an alias of admin.APIClient
type AdminAPIClient admin.APIClient

// TransactionAPIClient is an alias of transaction.APIClient
type TransactionAPIClient transaction.APIClient

// DebugClient is an alias of debug.DebugClient
type DebugClient debug.DebugClient

// ProxyClient is an alias of proxy.APIClient
type ProxyClient proxy.APIClient

// An APIClient is a wrapper around pfs, pps and block APIClients.
type APIClient struct {
	PfsAPIClient
	PpsAPIClient
	AuthAPIClient
	IdentityAPIClient
	VersionAPIClient
	AdminAPIClient
	TransactionAPIClient
	DebugClient
	ProxyClient
	Enterprise enterprise.APIClient // not embedded--method name conflicts with AuthAPIClient
	License    license.APIClient

	// addr is the parsed address used to connect to this server
	addr *grpcutil.PachdAddress

	// The trusted CAs, for authenticating a pachd server over TLS
	caCerts *x509.CertPool

	// gzipCompress configures whether to enable compression by default for all calls
	gzipCompress bool

	// clientConn is a cached grpc connection to 'addr'
	clientConn *grpc.ClientConn

	// healthClient is a cached healthcheck client connected to 'addr'
	healthClient grpc_health_v1.HealthClient

	// streamSemaphore limits the number of concurrent message streams between
	// this client and pachd
	limiter limit.ConcurrencyLimiter

	// metricsUserID is an identifier that is included in usage metrics sent to
	// Pachyderm Inc. and is used to count the number of unique Pachyderm users.
	// If unset, no usage metrics are sent back to Pachyderm Inc.
	metricsUserID string

	// metricsPrefix is used to send information from this client to Pachyderm Inc
	// for usage metrics
	metricsPrefix string

	// authenticationToken is an identifier that authenticates the caller in case
	// they want to access privileged data
	authenticationToken string

	// The context used in requests, can be set with WithCtx
	ctx context.Context

	portForwarder *PortForwarder

	// The client context name this client was created from, if it was created by
	// NewOnUserMachine
	clientContextName string

	defaultTransformImage string
	defaultTransformUser  string
}

// GetAddress returns the pachd host:port with which 'c' is communicating. If
// 'c' was created using NewInCluster or NewOnUserMachine then this is how the
// address may be retrieved from the environment.
func (c *APIClient) GetAddress() *grpcutil.PachdAddress {
	return c.addr
}

// DefaultMaxConcurrentStreams defines the max number of Putfiles or Getfiles happening simultaneously
const DefaultMaxConcurrentStreams = 100

// DefaultDialTimeout is the max amount of time APIClient.connect() will wait
// for a connection to be established unless overridden by WithDialTimeout()
const DefaultDialTimeout = 30 * time.Second

type clientSettings struct {
	maxConcurrentStreams int
	gzipCompress         bool
	dialTimeout          time.Duration
	caCerts              *x509.CertPool
	unaryInterceptors    []grpc.UnaryClientInterceptor
	streamInterceptors   []grpc.StreamClientInterceptor
}

// NewFromURI creates a new client given a GRPC URI ex. grpc://test.example.com.
// If no scheme is specified `grpc://` is assumed. A scheme of `grpcs://` enables TLS.
func NewFromURI(uri string, options ...Option) (*APIClient, error) {
	pachdAddress, err := grpcutil.ParsePachdAddress(uri)
	if err != nil {
		return nil, errors.Wrap(err, "could not parse the pachd address")
	}
	return NewFromPachdAddress(pachdAddress, options...)
}

// NewFromPachdAddress creates a new client given a parsed GRPC address
func NewFromPachdAddress(pachdAddress *grpcutil.PachdAddress, options ...Option) (*APIClient, error) {
	// By default, use the system CAs for secure connections
	// if no others are specified.
	if pachdAddress.Secured {
		options = append(options, WithSystemCAs)
	}

	// Apply creation options
	settings := clientSettings{
		maxConcurrentStreams: DefaultMaxConcurrentStreams,
		dialTimeout:          DefaultDialTimeout,
	}
	for _, option := range options {
		if err := option(&settings); err != nil {
			return nil, err
		}
	}
	if tracing.IsActive() {
		settings.unaryInterceptors = append(settings.unaryInterceptors, tracing.UnaryClientInterceptor())
		settings.streamInterceptors = append(settings.streamInterceptors, tracing.StreamClientInterceptor())
	}
	c := &APIClient{
		addr:         pachdAddress,
		caCerts:      settings.caCerts,
		limiter:      limit.New(settings.maxConcurrentStreams),
		gzipCompress: settings.gzipCompress,
	}
	if err := c.connect(settings.dialTimeout, settings.unaryInterceptors, settings.streamInterceptors); err != nil {
		return nil, err
	}
	return c, nil
}

// Option is a client creation option that may be passed to NewOnUserMachine(), or NewInCluster()
type Option func(*clientSettings) error

// WithMaxConcurrentStreams instructs the New* functions to create client that
// can have at most 'streams' concurrent streams open with pachd at a time
func WithMaxConcurrentStreams(streams int) Option {
	return func(settings *clientSettings) error {
		settings.maxConcurrentStreams = streams
		return nil
	}
}

func addCertFromFile(pool *x509.CertPool, path string) error {
	bytes, err := ioutil.ReadFile(path)
	if err != nil {
		return errors.Wrapf(err, "could not read x509 cert from \"%s\"", path)
	}
	if ok := pool.AppendCertsFromPEM(bytes); !ok {
		return errors.Errorf("could not add %s to cert pool as PEM", path)
	}
	return nil
}

// WithRootCAs instructs the New* functions to create client that uses the
// given signed x509 certificates as the trusted root certificates (instead of
// the system certs). Introduced to pass certs provided via command-line flags
func WithRootCAs(path string) Option {
	return func(settings *clientSettings) error {
		settings.caCerts = x509.NewCertPool()
		return addCertFromFile(settings.caCerts, path)
	}
}

// WithAdditionalRootCAs instructs the New* functions to additionally trust the
// given base64-encoded, signed x509 certificates as root certificates.
// Introduced to pass certs in the Pachyderm config
func WithAdditionalRootCAs(pemBytes []byte) Option {
	return func(settings *clientSettings) error {
		// append certs from config
		if settings.caCerts == nil {
			settings.caCerts = x509.NewCertPool()
		}
		if ok := settings.caCerts.AppendCertsFromPEM(pemBytes); !ok {
			return errors.Errorf("server CA certs are present in Pachyderm config, but could not be added to client")
		}
		return nil
	}
}

// WithSystemCAs uses the system certs for client creation, if no others are provided.
// This is the default behaviour when the scheme is `https` or `grpcs`.
func WithSystemCAs(settings *clientSettings) error {
	if settings.caCerts != nil {
		return nil
	}

	certs, err := x509.SystemCertPool()
	if err != nil {
		return errors.Wrap(err, "could not retrieve system cert pool")
	}
	settings.caCerts = certs
	return nil
}

// WithDialTimeout instructs the New* functions to use 't' as the deadline to
// connect to pachd
func WithDialTimeout(t time.Duration) Option {
	return func(settings *clientSettings) error {
		settings.dialTimeout = t
		return nil
	}
}

// WithGZIPCompression enabled GZIP compression for data on the wire
func WithGZIPCompression() Option {
	return func(settings *clientSettings) error {
		settings.gzipCompress = true
		return nil
	}
}

// WithAdditionalPachdCert instructs the New* functions to additionally trust
// the signed cert mounted in Pachd's cert volume. This is used by Pachd
// when connecting to itself (if no cert is present, the clients cert pool
// will not be modified, so that if no other options have been passed, pachd
// will connect to itself over an insecure connection)
func WithAdditionalPachdCert() Option {
	return func(settings *clientSettings) error {
		if _, err := os.Stat(tls.VolumePath); err == nil {
			if settings.caCerts == nil {
				settings.caCerts = x509.NewCertPool()
			}
			return addCertFromFile(settings.caCerts, path.Join(tls.VolumePath, tls.CertFile))
		}
		return nil
	}
}

// WithAdditionalUnaryClientInterceptors instructs the New* functions to add the provided
// UnaryClientInterceptors to the gRPC dial options when opening a client connection.  Internally,
// all of the provided options are coalesced into one chain, so it is safe to provide this option
// more than once.
//
// This client creates both Unary and Stream client connections, so you will probably want to supply
// a corresponding WithAdditionalStreamClientInterceptors call.
func WithAdditionalUnaryClientInterceptors(interceptors ...grpc.UnaryClientInterceptor) Option {
	return func(settings *clientSettings) error {
		settings.unaryInterceptors = append(settings.unaryInterceptors, interceptors...)
		return nil
	}
}

// WithAdditionalStreamClientInterceptors instructs the New* functions to add the provided
// StreamClientInterceptors to the gRPC dial options when opening a client connection.  Internally,
// all of the provided options are coalesced into one chain, so it is safe to provide this option
// more than once.
//
// This client creates both Unary and Stream client connections, so you will probably want to supply
// a corresponding WithAdditionalUnaryClientInterceptors option.
func WithAdditionalStreamClientInterceptors(interceptors ...grpc.StreamClientInterceptor) Option {
	return func(settings *clientSettings) error {
		settings.streamInterceptors = append(settings.streamInterceptors, interceptors...)
		return nil
	}
}

func getCertOptionsFromEnv() ([]Option, error) {
	var options []Option
	if certPaths, ok := os.LookupEnv("PACH_CA_CERTS"); ok {
		fmt.Fprintln(os.Stderr, "WARNING: 'PACH_CA_CERTS' is deprecated and will be removed in a future release, use Pachyderm contexts instead.")

		pachdAddressStr, ok := os.LookupEnv("PACHD_ADDRESS")
		if !ok {
			return nil, errors.New("cannot set 'PACH_CA_CERTS' without setting 'PACHD_ADDRESS'")
		}

		pachdAddress, err := grpcutil.ParsePachdAddress(pachdAddressStr)
		if err != nil {
			return nil, errors.Wrapf(err, "could not parse 'PACHD_ADDRESS'")
		}

		if !pachdAddress.Secured {
			return nil, errors.Errorf("cannot set 'PACH_CA_CERTS' if 'PACHD_ADDRESS' is not using grpcs")
		}

		paths := strings.Split(certPaths, ",")
		for _, p := range paths {
			// Try to read all certs under 'p'--skip any that we can't read/stat
			if err := filepath.Walk(p, func(p string, info os.FileInfo, err error) error {
				if err != nil {
					log.Warnf("skipping \"%s\", could not stat path: %v", p, err)
					return nil // Don't try and fix any errors encountered by Walk() itself
				}
				if info.IsDir() {
					return nil // We'll just read the children of any directories when we traverse them
				}
				pemBytes, err := ioutil.ReadFile(p)
				if err != nil {
					log.Warnf("could not read server CA certs at %s: %v", p, err)
					return nil
				}
				options = append(options, WithAdditionalRootCAs(pemBytes))
				return nil
			}); err != nil {
				return nil, err
			}
		}
	}
	return options, nil
}

// getUserMachineAddrAndOpts is a helper for NewOnUserMachine that uses
// environment variables, config files, etc to figure out which address a user
// running a command should connect to.
func getUserMachineAddrAndOpts(context *config.Context) (*grpcutil.PachdAddress, []Option, error) {
	var options []Option

	// 1) PACHD_ADDRESS environment variable (shell-local) overrides global config
	if envAddrStr, ok := os.LookupEnv("PACHD_ADDRESS"); ok {
		fmt.Fprintln(os.Stderr, "WARNING: 'PACHD_ADDRESS' is deprecated and will be removed in a future release, use Pachyderm contexts instead.")

		envAddr, err := grpcutil.ParsePachdAddress(envAddrStr)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "could not parse 'PACHD_ADDRESS'")
		}
		options, err := getCertOptionsFromEnv()
		if err != nil {
			return nil, nil, err
		}

		return envAddr, options, nil
	}

	// 2) Get target address from global config if possible
	if context != nil && (context.ServerCAs != "" || context.PachdAddress != "") {
		pachdAddress, err := grpcutil.ParsePachdAddress(context.PachdAddress)
		if err != nil {
			return nil, nil, errors.Wrap(err, "could not parse the active context's pachd address")
		}

		// Proactively return an error in this case, instead of falling back to the default address below
		if context.ServerCAs != "" && !pachdAddress.Secured {
			return nil, nil, errors.New("must set pachd_address to grpcs://... if server_cas is set")
		}

		if pachdAddress.Secured {
			options = append(options, WithSystemCAs)
		}
		// Also get cert info from config (if set)
		if context.ServerCAs != "" {
			pemBytes, err := base64.StdEncoding.DecodeString(context.ServerCAs)
			if err != nil {
				return nil, nil, errors.Wrap(err, "could not decode server CA certs in config")
			}
			return pachdAddress, []Option{WithAdditionalRootCAs(pemBytes)}, nil
		}
		return pachdAddress, options, nil
	}

	// 3) Use default address (broadcast) if nothing else works
	options, err := getCertOptionsFromEnv() // error if PACH_CA_CERTS is set
	if err != nil {
		return nil, nil, err
	}
	return nil, options, nil
}

func portForwarder(context *config.Context) (*PortForwarder, uint16, error) {
	fw, err := NewPortForwarder(context, "")
	if err != nil {
		return nil, 0, errors.Wrap(err, "failed to initialize port forwarder")
	}

	var port uint16
	if context.EnterpriseServer {
		port, err = fw.RunForEnterpriseServer(0, 1650)
	} else {
		port, err = fw.RunForPachd(0, 1650)
	}

	if err != nil {
		return nil, 0, err
	}
	log.Debugf("Implicit port forwarder listening on port %d", port)

	return fw, port, nil
}

// NewForTest constructs a new APIClient for tests.
// TODO(actgardner): this should probably live in testutils and accept a testing.TB
func NewForTest() (*APIClient, error) {
	cfg, err := config.Read(false, false)
	if err != nil {
		return nil, errors.Wrap(err, "could not read config")
	}
	_, context, err := cfg.ActiveContext(true)
	if err != nil {
		return nil, errors.Wrap(err, "could not get active context")
	}

	// create new pachctl client
	pachdAddress, cfgOptions, err := getUserMachineAddrAndOpts(context)
	if err != nil {
		return nil, err
	}

	if pachdAddress == nil {
		pachdAddress = &grpcutil.DefaultPachdAddress
	}

	client, err := NewFromPachdAddress(pachdAddress, cfgOptions...)
	if err != nil {
		return nil, errors.Wrapf(err, "could not connect to pachd at %s", pachdAddress.Qualified())
	}
	return client, nil
}

// NewEnterpriseClientForTest constructs a new APIClient for tests.
// TODO(actgardner): this should probably live in testutils and accept a testing.TB
func NewEnterpriseClientForTest() (*APIClient, error) {
	cfg, err := config.Read(false, false)
	if err != nil {
		return nil, errors.Wrap(err, "could not read config")
	}
	_, context, err := cfg.ActiveEnterpriseContext(true)
	if err != nil {
		return nil, errors.Wrap(err, "could not get active context")
	}

	// create new pachctl client
	pachdAddress, cfgOptions, err := getUserMachineAddrAndOpts(context)
	if err != nil {
		return nil, err
	}

	if pachdAddress == nil {
		return nil, errors.New("no enterprise server configured")
	}

	client, err := NewFromPachdAddress(pachdAddress, cfgOptions...)
	if err != nil {
		return nil, errors.Wrapf(err, "could not connect to pachd at %s", pachdAddress.Qualified())
	}
	return client, nil
}

// NewOnUserMachine constructs a new APIClient using $HOME/.pachyderm/config
// if it exists. This is intended to be used in the pachctl binary.
func NewOnUserMachine(prefix string, options ...Option) (*APIClient, error) {
	cfg, err := config.Read(false, false)
	if err != nil {
		return nil, errors.Wrap(err, "could not read config")
	}
	name, context, err := cfg.ActiveContext(true)
	if err != nil {
		return nil, errors.Wrap(err, "could not get active context")
	}
	return newOnUserMachine(cfg, context, name, prefix, options...)
}

// NewEnterpriseClientOnUserMachine constructs a new APIClient using $HOME/.pachyderm/config
// if it exists. This is intended to be used in the pachctl binary to communicate with the
// enterprise server.
func NewEnterpriseClientOnUserMachine(prefix string, options ...Option) (*APIClient, error) {
	cfg, err := config.Read(false, false)
	if err != nil {
		return nil, errors.Wrap(err, "could not read config")
	}
	name, context, err := cfg.ActiveEnterpriseContext(true)
	if err != nil {
		return nil, errors.Wrap(err, "could not get active context")
	}
	return newOnUserMachine(cfg, context, name, prefix, options...)
}

// TODO(msteffen) this logic is fairly linux/unix specific, and makes the
// pachyderm client library incompatible with Windows. We may want to move this
// (and similar) logic into src/server and have it call a NewFromOptions()
// constructor.
func newOnUserMachine(cfg *config.Config, context *config.Context, contextName, prefix string, options ...Option) (*APIClient, error) {
	// create new pachctl client
	pachdAddress, cfgOptions, err := getUserMachineAddrAndOpts(context)
	if err != nil {
		return nil, err
	}

	var fw *PortForwarder
	if pachdAddress == nil && context.PortForwarders != nil {
		pachdLocalPort, ok := context.PortForwarders["pachd"]
		if ok {
			log.Debugf("Connecting to explicitly port forwarded pachd instance on port %d", pachdLocalPort)
			pachdAddress = &grpcutil.PachdAddress{
				Secured: false,
				Host:    "localhost",
				Port:    uint16(pachdLocalPort),
			}
		}
	}
	if pachdAddress == nil {
		var pachdLocalPort uint16
		fw, pachdLocalPort, err = portForwarder(context)
		if err != nil {
			return nil, err
		}
		pachdAddress = &grpcutil.PachdAddress{
			Secured: false,
			Host:    "localhost",
			Port:    pachdLocalPort,
		}
	}

	client, err := NewFromPachdAddress(pachdAddress, append(options, cfgOptions...)...)
	if err != nil {
		return nil, errors.Wrapf(err, "could not connect to pachd at %q", pachdAddress.Qualified())
	}

	// Add metrics info & authentication token
	client.metricsPrefix = prefix
	if cfg.UserID != "" && cfg.V2.Metrics {
		client.metricsUserID = cfg.UserID
	}
	if context.SessionToken != "" {
		client.authenticationToken = context.SessionToken
	}

	// Verify cluster deployment ID
	clusterInfo, err := client.InspectCluster()
	if err != nil {
		if strings.Contains("unknown service admin_v2.API", err.Error()) {
			return nil, fmt.Errorf("this client is for pachyderm 2.x, but the server has a different version - please install the correct client for your server")

		}
		return nil, errors.Wrap(err, "could not get cluster ID")
	}
	if context.ClusterDeploymentID != clusterInfo.DeploymentID {
		if context.ClusterDeploymentID == "" {
			context.ClusterDeploymentID = clusterInfo.DeploymentID
			if err = cfg.Write(); err != nil {
				return nil, errors.Wrap(err, "could not write config to save cluster deployment ID")
			}
		} else {
			return nil, errors.Errorf("connected to the wrong cluster (context cluster deployment ID = %q vs reported cluster deployment ID = %q)", context.ClusterDeploymentID, clusterInfo.DeploymentID)
		}
	}

	// Add port forwarding. This will set it to nil if port forwarding is
	// disabled, or an address is explicitly set.
	client.portForwarder = fw

	client.clientContextName = contextName

	return client, nil
}

// NewInCluster constructs a new APIClient using env vars that Kubernetes creates.
// This should be used to access Pachyderm from within a Kubernetes cluster
// with Pachyderm running on it.
func NewInCluster(options ...Option) (*APIClient, error) {
	// first try the pachd peer service (only supported on pachyderm >= 1.10),
	// which will work when TLS is enabled
	internalHost := os.Getenv("PACHD_PEER_SERVICE_HOST")
	internalPort := os.Getenv("PACHD_PEER_SERVICE_PORT")
	if internalHost != "" && internalPort != "" {
		return NewFromURI(fmt.Sprintf("%s:%s", internalHost, internalPort), options...)
	}

	host, ok := os.LookupEnv("PACHD_SERVICE_HOST")
	if !ok {
		return nil, errors.Errorf("PACHD_SERVICE_HOST not set")
	}
	port, ok := os.LookupEnv("PACHD_SERVICE_PORT")
	if !ok {
		return nil, errors.Errorf("PACHD_SERVICE_PORT not set")
	}
	// create new pachctl client
	return NewFromURI(fmt.Sprintf("%s:%s", host, port), options...)
}

// NewInWorker constructs a new APIClient intended to be used from a worker
// to talk to the sidecar pachd container
func NewInWorker(options ...Option) (*APIClient, error) {
	cfg, err := config.Read(false, true)
	if err != nil {
		return nil, errors.Wrap(err, "could not read config")
	}
	_, context, err := cfg.ActiveContext(true)
	if err != nil {
		return nil, errors.Wrap(err, "could not get active context")
	}

	if localPort, ok := os.LookupEnv("PEER_PORT"); ok {
		client, err := NewFromURI(fmt.Sprintf("127.0.0.1:%s", localPort), options...)
		if err != nil {
			return nil, errors.Wrap(err, "could not create client")
		}
		if context.SessionToken != "" {
			client.authenticationToken = context.SessionToken
		}
		return client, nil
	}
	return nil, errors.New("PEER_PORT not set")
}

// Close the connection to gRPC
func (c *APIClient) Close() error {
	if err := c.clientConn.Close(); err != nil {
		return err
	}

	if c.portForwarder != nil {
		c.portForwarder.Close()
	}

	return nil
}

// DeleteAll deletes everything in the cluster.
// Use with caution, there is no undo.
// TODO: rewrite this to use transactions
func (c APIClient) DeleteAll() error {
	if _, err := c.IdentityAPIClient.DeleteAll(
		c.Ctx(),
		&identity.DeleteAllRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.AuthAPIClient.Deactivate(
		c.Ctx(),
		&auth.DeactivateRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.License.DeleteAll(
		c.Ctx(),
		&license.DeleteAllRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.PpsAPIClient.DeleteAll(
		c.Ctx(),
		&types.Empty{},
	); err != nil {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.PfsAPIClient.DeleteAll(
		c.Ctx(),
		&types.Empty{},
	); err != nil {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.TransactionAPIClient.DeleteAll(
		c.Ctx(),
		&transaction.DeleteAllRequest{},
	); err != nil {
		return grpcutil.ScrubGRPC(err)
	}
	return nil
}

// DeleteAllEnterprise deletes everything in the enterprise server.
// Use with caution, there is no undo.
// TODO: rewrite this to use transactions
func (c APIClient) DeleteAllEnterprise() error {
	if _, err := c.IdentityAPIClient.DeleteAll(
		c.Ctx(),
		&identity.DeleteAllRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.AuthAPIClient.Deactivate(
		c.Ctx(),
		&auth.DeactivateRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	if _, err := c.License.DeleteAll(
		c.Ctx(),
		&license.DeleteAllRequest{},
	); err != nil && !auth.IsErrNotActivated(err) {
		return grpcutil.ScrubGRPC(err)
	}
	return nil
}

// SetMaxConcurrentStreams Sets the maximum number of concurrent streams the
// client can have. It is not safe to call this operations while operations are
// outstanding.
func (c APIClient) SetMaxConcurrentStreams(n int) {
	c.limiter = limit.New(n)
}

// DefaultDialOptions is a helper returning a slice of grpc.Dial options
// such that grpc.Dial() is synchronous: the call doesn't return until
// the connection has been established and it's safe to send RPCs
func DefaultDialOptions() []grpc.DialOption {
	return []grpc.DialOption{
		// Don't return from Dial() until the connection has been established.
		grpc.WithBlock(),
		grpc.WithKeepaliveParams(keepalive.ClientParameters{
			Time:                20 * time.Second,
			Timeout:             20 * time.Second,
			PermitWithoutStream: true,
		}),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallRecvMsgSize(grpcutil.MaxMsgSize),
			grpc.MaxCallSendMsgSize(grpcutil.MaxMsgSize),
		),
	}
}

func (c *APIClient) connect(timeout time.Duration, unaryInterceptors []grpc.UnaryClientInterceptor, streamInterceptors []grpc.StreamClientInterceptor) error {
	dialOptions := DefaultDialOptions()
	if c.caCerts == nil {
		dialOptions = append(dialOptions, grpc.WithInsecure())
	} else {
		tlsCreds := credentials.NewClientTLSFromCert(c.caCerts, "")
		dialOptions = append(dialOptions, grpc.WithTransportCredentials(tlsCreds))
	}
	if c.gzipCompress {
		dialOptions = append(dialOptions, grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
	}
	if len(unaryInterceptors) > 0 {
		dialOptions = append(dialOptions, grpc.WithChainUnaryInterceptor(unaryInterceptors...))
	}
	if len(streamInterceptors) > 0 {
		dialOptions = append(dialOptions, grpc.WithChainStreamInterceptor(streamInterceptors...))
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	// By default GRPC will attempt to get service config from a TXT record when
	// the `dns:///` scheme is used. Some DNS servers return the wrong type of error
	// when the TXT record doesn't exist, and it causes GRPC to back off and retry
	// service discovery forever.
	dialOptions = append(dialOptions, grpc.WithDisableServiceConfig())

	clientConn, err := grpc.DialContext(ctx, c.addr.Target(), dialOptions...)
	if err != nil {
		return err
	}
	c.PfsAPIClient = pfs.NewAPIClient(clientConn)
	c.PpsAPIClient = pps.NewAPIClient(clientConn)
	c.AuthAPIClient = auth.NewAPIClient(clientConn)
	c.IdentityAPIClient = identity.NewAPIClient(clientConn)
	c.Enterprise = enterprise.NewAPIClient(clientConn)
	c.License = license.NewAPIClient(clientConn)
	c.VersionAPIClient = versionpb.NewAPIClient(clientConn)
	c.AdminAPIClient = admin.NewAPIClient(clientConn)
	c.TransactionAPIClient = transaction.NewAPIClient(clientConn)
	c.DebugClient = debug.NewDebugClient(clientConn)
	c.ProxyClient = proxy.NewAPIClient(clientConn)
	c.clientConn = clientConn
	c.healthClient = grpc_health_v1.NewHealthClient(clientConn)
	return nil
}

// AddMetadata adds necessary metadata (including authentication credentials)
// to the context 'ctx', preserving any metadata that is present in either the
// incoming or outgoing metadata of 'ctx'.
func (c *APIClient) AddMetadata(ctx context.Context) context.Context {
	// TODO(msteffen): There are several places in this client where it's possible
	// to set per-request metadata (specifically auth tokens): client.WithCtx(),
	// client.SetAuthToken(), etc. These should be consolidated, as this API
	// doesn't make it obvious how these settings are resolved when they conflict.
	clientData := make(map[string]string)
	if c.authenticationToken != "" {
		clientData[auth.ContextTokenKey] = c.authenticationToken
	}
	// metadata API downcases all the key names
	if c.metricsUserID != "" {
		clientData["userid"] = c.metricsUserID
		clientData["prefix"] = c.metricsPrefix
	}

	// Rescue any metadata pairs already in 'ctx' (otherwise
	// metadata.NewOutgoingContext() would drop them). Note that this is similar
	// to metadata.Join(), but distinct because it discards conflicting k/v pairs
	// instead of merging them)
	incomingMD, _ := metadata.FromIncomingContext(ctx)
	outgoingMD, _ := metadata.FromOutgoingContext(ctx)
	clientMD := metadata.New(clientData)
	finalMD := make(metadata.MD) // Collect k/v pairs
	for _, md := range []metadata.MD{incomingMD, outgoingMD, clientMD} {
		for k, v := range md {
			finalMD[k] = v
		}
	}
	return metadata.NewOutgoingContext(ctx, finalMD)
}

// Ctx is a convenience function that returns adds Pachyderm authn metadata
// to context.Background().
func (c *APIClient) Ctx() context.Context {
	if c.ctx == nil {
		return c.AddMetadata(context.Background())
	}
	return c.AddMetadata(c.ctx)
}

// WithCtx returns a new APIClient that uses ctx for requests it sends. Note
// that the new APIClient will still use the authentication token and metrics
// metadata of this client, so this is only useful for propagating other
// context-associated metadata.
func (c *APIClient) WithCtx(ctx context.Context) *APIClient {
	result := *c // copy c
	result.ctx = ctx
	return &result
}

// AuthToken gets the authentication token that is set for this client.
func (c *APIClient) AuthToken() string {
	return c.authenticationToken
}

// SetAuthToken sets the authentication token that will be used for all
// API calls for this client.
func (c *APIClient) SetAuthToken(token string) {
	c.authenticationToken = token
}

// ClientContextName returns the name of the context in the client config
// that produced this client, or an empty string if the client was not
// produced from a configured client context.
func (c *APIClient) ClientContextName() string {
	return c.clientContextName
}


text:
---
layout: post
title: Leetcode Problem 190 Summary
date: 2016-08-17
categories: blog
tags: [study]

---

# 题目

The API: int read4(char *buf) reads 4 characters at a time from a file.

The return value is the actual number of characters read. For example, it returns 3 if there is only 3 characters left in the file.

By using the read4 API, implement the function int read(char *buf, int n) that reads n characters from the file.

Note:
The read function will only be called once for each test case.

# 我的算法

主要考察边界条件的考虑,以及能否熟练应用新的API吧。

# 代码

{% highlight java %}
/* The read4 API is defined in the parent class Reader4.
      int read4(char[] buf); */

public class Solution extends Reader4 {
    /**
     * @param buf Destination buffer
     * @param n   Maximum number of characters to read
     * @return    The number of characters read
     */
    public int read(char[] buf, int n) {
        int remain = n;
        int iter = 0;

        while (remain > 0) {
            char[] read4Buff = new char[4];
            int len = read4(read4Buff);
            for (int i = 0; i < len; i++) {
                buf[iter++] = read4Buff[i];
                if (iter == n) return iter;
            }
            if (len < 4) return iter;
        }

        return iter;
    }
}
{% endhighlight %}


text:
using Indulged.API.Storage.Models;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Indulged.API.Utils;

namespace Indulged.API.Storage.Factories
{
    public class TopicReplyFactory
    {
        public static FlickrReply TopicReplyWithJObject(JObject json, FlickrTopic topic)
        {
            // Get reply id
            string replyId = json["id"].ToString();
            FlickrReply reply = null;
            if (topic.ReplyCache.ContainsKey(replyId))
            {
                reply = topic.ReplyCache[replyId];
            }
            else
            {
                reply = new FlickrReply();
                reply.ResourceId = replyId;
                topic.ReplyCache[replyId] = reply;
            }

            // Parse user
            reply.Author = FlickrUserFactory.UserWithTopicJObject(json);
            reply.IsAdmin = (json["role"].ToString() == "admin");
            reply.CreationDate = json["datecreate"].ToString().ToDateTime();

            // Message
            reply.Message = json["message"]["_content"].ToString();

            // Replies
            reply.CanDelete = (json["can_delete"].ToString() == "1");

            return reply;
        }
    }
}


text:
Detects if operation on collection may be simplified.

Before:

option.map(myValueIsGoodEnough).getOrElse(false)
sequence.find(function).isDefined
iterable.foldLeft(0) { _ + _ }

After:

option.exists(goodEnoughValue)
sequence.exists(function)
iterable.sum


text:
/*
 * Cross Partition Communication (XPC) uv-based functions.
 *
 *     Architecture specific implementation of common functions.
 *
 */

#include <linux/kernel.h>
#include <linux/mm.h>
#include <linux/interrupt.h>
#include <linux/delay.h>
#include <linux/device.h>
#include <linux/cpu.h>
#include <linux/module.h>
#include <linux/err.h>
#include <linux/slab.h>
#include <asm/uv/uv_hub.h>
#if defined CONFIG_X86_64
#include <asm/uv/bios.h>
#include <asm/uv/uv_irq.h>
#elif defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
#include <asm/sn/intr.h>
#include <asm/sn/sn_sal.h>
#endif
#include "../sgi-gru/gru.h"
#include "../sgi-gru/grukservices.h"
#include "xpc.h"

#if defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
struct uv_IO_APIC_route_entry {
	__u64	vector		:  8,
		delivery_mode	:  3,
		dest_mode	:  1,
		delivery_status	:  1,
		polarity	:  1,
		__reserved_1	:  1,
		trigger		:  1,
		mask		:  1,
		__reserved_2	: 15,
		dest		: 32;
};
#endif

static struct xpc_heartbeat_uv *xpc_heartbeat_uv;

#define XPC_ACTIVATE_MSG_SIZE_UV	(1 * GRU_CACHE_LINE_BYTES)
#define XPC_ACTIVATE_MQ_SIZE_UV		(4 * XP_MAX_NPARTITIONS_UV * \
					 XPC_ACTIVATE_MSG_SIZE_UV)
#define XPC_ACTIVATE_IRQ_NAME		"xpc_activate"

#define XPC_NOTIFY_MSG_SIZE_UV		(2 * GRU_CACHE_LINE_BYTES)
#define XPC_NOTIFY_MQ_SIZE_UV		(4 * XP_MAX_NPARTITIONS_UV * \
					 XPC_NOTIFY_MSG_SIZE_UV)
#define XPC_NOTIFY_IRQ_NAME		"xpc_notify"

static int xpc_mq_node = -1;

static struct xpc_gru_mq_uv *xpc_activate_mq_uv;
static struct xpc_gru_mq_uv *xpc_notify_mq_uv;

static int
xpc_setup_partitions_uv(void)
{
	short partid;
	struct xpc_partition_uv *part_uv;

	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
		part_uv = &xpc_partitions[partid].sn.uv;

		mutex_init(&part_uv->cached_activate_gru_mq_desc_mutex);
		spin_lock_init(&part_uv->flags_lock);
		part_uv->remote_act_state = XPC_P_AS_INACTIVE;
	}
	return 0;
}

static void
xpc_teardown_partitions_uv(void)
{
	short partid;
	struct xpc_partition_uv *part_uv;
	unsigned long irq_flags;

	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
		part_uv = &xpc_partitions[partid].sn.uv;

		if (part_uv->cached_activate_gru_mq_desc != NULL) {
			mutex_lock(&part_uv->cached_activate_gru_mq_desc_mutex);
			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
			part_uv->flags &= ~XPC_P_CACHED_ACTIVATE_GRU_MQ_DESC_UV;
			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
			kfree(part_uv->cached_activate_gru_mq_desc);
			part_uv->cached_activate_gru_mq_desc = NULL;
			mutex_unlock(&part_uv->
				     cached_activate_gru_mq_desc_mutex);
		}
	}
}

static int
xpc_get_gru_mq_irq_uv(struct xpc_gru_mq_uv *mq, int cpu, char *irq_name)
{
	int mmr_pnode = uv_blade_to_pnode(mq->mmr_blade);

#if defined CONFIG_X86_64
	mq->irq = uv_setup_irq(irq_name, cpu, mq->mmr_blade, mq->mmr_offset,
			UV_AFFINITY_CPU);
	if (mq->irq < 0)
		return mq->irq;

	mq->mmr_value = uv_read_global_mmr64(mmr_pnode, mq->mmr_offset);

#elif defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
	if (strcmp(irq_name, XPC_ACTIVATE_IRQ_NAME) == 0)
		mq->irq = SGI_XPC_ACTIVATE;
	else if (strcmp(irq_name, XPC_NOTIFY_IRQ_NAME) == 0)
		mq->irq = SGI_XPC_NOTIFY;
	else
		return -EINVAL;

	mq->mmr_value = (unsigned long)cpu_physical_id(cpu) << 32 | mq->irq;
	uv_write_global_mmr64(mmr_pnode, mq->mmr_offset, mq->mmr_value);
#else
	#error not a supported configuration
#endif

	return 0;
}

static void
xpc_release_gru_mq_irq_uv(struct xpc_gru_mq_uv *mq)
{
#if defined CONFIG_X86_64
	uv_teardown_irq(mq->irq);

#elif defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
	int mmr_pnode;
	unsigned long mmr_value;

	mmr_pnode = uv_blade_to_pnode(mq->mmr_blade);
	mmr_value = 1UL << 16;

	uv_write_global_mmr64(mmr_pnode, mq->mmr_offset, mmr_value);
#else
	#error not a supported configuration
#endif
}

static int
xpc_gru_mq_watchlist_alloc_uv(struct xpc_gru_mq_uv *mq)
{
	int ret;

#if defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
	int mmr_pnode = uv_blade_to_pnode(mq->mmr_blade);

	ret = sn_mq_watchlist_alloc(mmr_pnode, (void *)uv_gpa(mq->address),
				    mq->order, &mq->mmr_offset);
	if (ret < 0) {
		dev_err(xpc_part, "sn_mq_watchlist_alloc() failed, ret=%d\n",
			ret);
		return -EBUSY;
	}
#elif defined CONFIG_X86_64
	ret = uv_bios_mq_watchlist_alloc(uv_gpa(mq->address),
					 mq->order, &mq->mmr_offset);
	if (ret < 0) {
		dev_err(xpc_part, "uv_bios_mq_watchlist_alloc() failed, "
			"ret=%d\n", ret);
		return ret;
	}
#else
	#error not a supported configuration
#endif

	mq->watchlist_num = ret;
	return 0;
}

static void
xpc_gru_mq_watchlist_free_uv(struct xpc_gru_mq_uv *mq)
{
	int ret;
	int mmr_pnode = uv_blade_to_pnode(mq->mmr_blade);

#if defined CONFIG_X86_64
	ret = uv_bios_mq_watchlist_free(mmr_pnode, mq->watchlist_num);
	BUG_ON(ret != BIOS_STATUS_SUCCESS);
#elif defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
	ret = sn_mq_watchlist_free(mmr_pnode, mq->watchlist_num);
	BUG_ON(ret != SALRET_OK);
#else
	#error not a supported configuration
#endif
}

static struct xpc_gru_mq_uv *
xpc_create_gru_mq_uv(unsigned int mq_size, int cpu, char *irq_name,
		     irq_handler_t irq_handler)
{
	enum xp_retval xp_ret;
	int ret;
	int nid;
	int nasid;
	int pg_order;
	struct page *page;
	struct xpc_gru_mq_uv *mq;
	struct uv_IO_APIC_route_entry *mmr_value;

	mq = kmalloc(sizeof(struct xpc_gru_mq_uv), GFP_KERNEL);
	if (mq == NULL) {
		dev_err(xpc_part, "xpc_create_gru_mq_uv() failed to kmalloc() "
			"a xpc_gru_mq_uv structure\n");
		ret = -ENOMEM;
		goto out_0;
	}

	mq->gru_mq_desc = kzalloc(sizeof(struct gru_message_queue_desc),
				  GFP_KERNEL);
	if (mq->gru_mq_desc == NULL) {
		dev_err(xpc_part, "xpc_create_gru_mq_uv() failed to kmalloc() "
			"a gru_message_queue_desc structure\n");
		ret = -ENOMEM;
		goto out_1;
	}

	pg_order = get_order(mq_size);
	mq->order = pg_order + PAGE_SHIFT;
	mq_size = 1UL << mq->order;

	mq->mmr_blade = uv_cpu_to_blade_id(cpu);

	nid = cpu_to_node(cpu);
	page = alloc_pages_exact_node(nid,
				      GFP_KERNEL | __GFP_ZERO | GFP_THISNODE,
				      pg_order);
	if (page == NULL) {
		dev_err(xpc_part, "xpc_create_gru_mq_uv() failed to alloc %d "
			"bytes of memory on nid=%d for GRU mq\n", mq_size, nid);
		ret = -ENOMEM;
		goto out_2;
	}
	mq->address = page_address(page);

	/* enable generation of irq when GRU mq operation occurs to this mq */
	ret = xpc_gru_mq_watchlist_alloc_uv(mq);
	if (ret != 0)
		goto out_3;

	ret = xpc_get_gru_mq_irq_uv(mq, cpu, irq_name);
	if (ret != 0)
		goto out_4;

	ret = request_irq(mq->irq, irq_handler, 0, irq_name, NULL);
	if (ret != 0) {
		dev_err(xpc_part, "request_irq(irq=%d) returned error=%d\n",
			mq->irq, -ret);
		goto out_5;
	}

	nasid = UV_PNODE_TO_NASID(uv_cpu_to_pnode(cpu));

	mmr_value = (struct uv_IO_APIC_route_entry *)&mq->mmr_value;
	ret = gru_create_message_queue(mq->gru_mq_desc, mq->address, mq_size,
				     nasid, mmr_value->vector, mmr_value->dest);
	if (ret != 0) {
		dev_err(xpc_part, "gru_create_message_queue() returned "
			"error=%d\n", ret);
		ret = -EINVAL;
		goto out_6;
	}

	/* allow other partitions to access this GRU mq */
	xp_ret = xp_expand_memprotect(xp_pa(mq->address), mq_size);
	if (xp_ret != xpSuccess) {
		ret = -EACCES;
		goto out_6;
	}

	return mq;

	/* something went wrong */
out_6:
	free_irq(mq->irq, NULL);
out_5:
	xpc_release_gru_mq_irq_uv(mq);
out_4:
	xpc_gru_mq_watchlist_free_uv(mq);
out_3:
	free_pages((unsigned long)mq->address, pg_order);
out_2:
	kfree(mq->gru_mq_desc);
out_1:
	kfree(mq);
out_0:
	return ERR_PTR(ret);
}

static void
xpc_destroy_gru_mq_uv(struct xpc_gru_mq_uv *mq)
{
	unsigned int mq_size;
	int pg_order;
	int ret;

	/* disallow other partitions to access GRU mq */
	mq_size = 1UL << mq->order;
	ret = xp_restrict_memprotect(xp_pa(mq->address), mq_size);
	BUG_ON(ret != xpSuccess);

	/* unregister irq handler and release mq irq/vector mapping */
	free_irq(mq->irq, NULL);
	xpc_release_gru_mq_irq_uv(mq);

	/* disable generation of irq when GRU mq op occurs to this mq */
	xpc_gru_mq_watchlist_free_uv(mq);

	pg_order = mq->order - PAGE_SHIFT;
	free_pages((unsigned long)mq->address, pg_order);

	kfree(mq);
}

static enum xp_retval
xpc_send_gru_msg(struct gru_message_queue_desc *gru_mq_desc, void *msg,
		 size_t msg_size)
{
	enum xp_retval xp_ret;
	int ret;

	while (1) {
		ret = gru_send_message_gpa(gru_mq_desc, msg, msg_size);
		if (ret == MQE_OK) {
			xp_ret = xpSuccess;
			break;
		}

		if (ret == MQE_QUEUE_FULL) {
			dev_dbg(xpc_chan, "gru_send_message_gpa() returned "
				"error=MQE_QUEUE_FULL\n");
			/* !!! handle QLimit reached; delay & try again */
			/* ??? Do we add a limit to the number of retries? */
			(void)msleep_interruptible(10);
		} else if (ret == MQE_CONGESTION) {
			dev_dbg(xpc_chan, "gru_send_message_gpa() returned "
				"error=MQE_CONGESTION\n");
			/* !!! handle LB Overflow; simply try again */
			/* ??? Do we add a limit to the number of retries? */
		} else {
			/* !!! Currently this is MQE_UNEXPECTED_CB_ERR */
			dev_err(xpc_chan, "gru_send_message_gpa() returned "
				"error=%d\n", ret);
			xp_ret = xpGruSendMqError;
			break;
		}
	}
	return xp_ret;
}

static void
xpc_process_activate_IRQ_rcvd_uv(void)
{
	unsigned long irq_flags;
	short partid;
	struct xpc_partition *part;
	u8 act_state_req;

	DBUG_ON(xpc_activate_IRQ_rcvd == 0);

	spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
		part = &xpc_partitions[partid];

		if (part->sn.uv.act_state_req == 0)
			continue;

		xpc_activate_IRQ_rcvd--;
		BUG_ON(xpc_activate_IRQ_rcvd < 0);

		act_state_req = part->sn.uv.act_state_req;
		part->sn.uv.act_state_req = 0;
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		if (act_state_req == XPC_P_ASR_ACTIVATE_UV) {
			if (part->act_state == XPC_P_AS_INACTIVE)
				xpc_activate_partition(part);
			else if (part->act_state == XPC_P_AS_DEACTIVATING)
				XPC_DEACTIVATE_PARTITION(part, xpReactivating);

		} else if (act_state_req == XPC_P_ASR_REACTIVATE_UV) {
			if (part->act_state == XPC_P_AS_INACTIVE)
				xpc_activate_partition(part);
			else
				XPC_DEACTIVATE_PARTITION(part, xpReactivating);

		} else if (act_state_req == XPC_P_ASR_DEACTIVATE_UV) {
			XPC_DEACTIVATE_PARTITION(part, part->sn.uv.reason);

		} else {
			BUG();
		}

		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (xpc_activate_IRQ_rcvd == 0)
			break;
	}
	spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

}

static void
xpc_handle_activate_mq_msg_uv(struct xpc_partition *part,
			      struct xpc_activate_mq_msghdr_uv *msg_hdr,
			      int part_setup,
			      int *wakeup_hb_checker)
{
	unsigned long irq_flags;
	struct xpc_partition_uv *part_uv = &part->sn.uv;
	struct xpc_openclose_args *args;

	part_uv->remote_act_state = msg_hdr->act_state;

	switch (msg_hdr->type) {
	case XPC_ACTIVATE_MQ_MSG_SYNC_ACT_STATE_UV:
		/* syncing of remote_act_state was just done above */
		break;

	case XPC_ACTIVATE_MQ_MSG_ACTIVATE_REQ_UV: {
		struct xpc_activate_mq_msg_activate_req_uv *msg;

		/*
		 * ??? Do we deal here with ts_jiffies being different
		 * ??? if act_state != XPC_P_AS_INACTIVE instead of
		 * ??? below?
		 */
		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_activate_req_uv, hdr);

		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (part_uv->act_state_req == 0)
			xpc_activate_IRQ_rcvd++;
		part_uv->act_state_req = XPC_P_ASR_ACTIVATE_UV;
		part->remote_rp_pa = msg->rp_gpa; /* !!! _pa is _gpa */
		part->remote_rp_ts_jiffies = msg_hdr->rp_ts_jiffies;
		part_uv->heartbeat_gpa = msg->heartbeat_gpa;

		if (msg->activate_gru_mq_desc_gpa !=
		    part_uv->activate_gru_mq_desc_gpa) {
			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
			part_uv->flags &= ~XPC_P_CACHED_ACTIVATE_GRU_MQ_DESC_UV;
			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
			part_uv->activate_gru_mq_desc_gpa =
			    msg->activate_gru_mq_desc_gpa;
		}
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		(*wakeup_hb_checker)++;
		break;
	}
	case XPC_ACTIVATE_MQ_MSG_DEACTIVATE_REQ_UV: {
		struct xpc_activate_mq_msg_deactivate_req_uv *msg;

		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_deactivate_req_uv, hdr);

		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (part_uv->act_state_req == 0)
			xpc_activate_IRQ_rcvd++;
		part_uv->act_state_req = XPC_P_ASR_DEACTIVATE_UV;
		part_uv->reason = msg->reason;
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		(*wakeup_hb_checker)++;
		return;
	}
	case XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREQUEST_UV: {
		struct xpc_activate_mq_msg_chctl_closerequest_uv *msg;

		if (!part_setup)
			break;

		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_chctl_closerequest_uv,
				   hdr);
		args = &part->remote_openclose_args[msg->ch_number];
		args->reason = msg->reason;

		spin_lock_irqsave(&part->chctl_lock, irq_flags);
		part->chctl.flags[msg->ch_number] |= XPC_CHCTL_CLOSEREQUEST;
		spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

		xpc_wakeup_channel_mgr(part);
		break;
	}
	case XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREPLY_UV: {
		struct xpc_activate_mq_msg_chctl_closereply_uv *msg;

		if (!part_setup)
			break;

		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_chctl_closereply_uv,
				   hdr);

		spin_lock_irqsave(&part->chctl_lock, irq_flags);
		part->chctl.flags[msg->ch_number] |= XPC_CHCTL_CLOSEREPLY;
		spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

		xpc_wakeup_channel_mgr(part);
		break;
	}
	case XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREQUEST_UV: {
		struct xpc_activate_mq_msg_chctl_openrequest_uv *msg;

		if (!part_setup)
			break;

		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_chctl_openrequest_uv,
				   hdr);
		args = &part->remote_openclose_args[msg->ch_number];
		args->entry_size = msg->entry_size;
		args->local_nentries = msg->local_nentries;

		spin_lock_irqsave(&part->chctl_lock, irq_flags);
		part->chctl.flags[msg->ch_number] |= XPC_CHCTL_OPENREQUEST;
		spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

		xpc_wakeup_channel_mgr(part);
		break;
	}
	case XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREPLY_UV: {
		struct xpc_activate_mq_msg_chctl_openreply_uv *msg;

		if (!part_setup)
			break;

		msg = container_of(msg_hdr, struct
				   xpc_activate_mq_msg_chctl_openreply_uv, hdr);
		args = &part->remote_openclose_args[msg->ch_number];
		args->remote_nentries = msg->remote_nentries;
		args->local_nentries = msg->local_nentries;
		args->local_msgqueue_pa = msg->notify_gru_mq_desc_gpa;

		spin_lock_irqsave(&part->chctl_lock, irq_flags);
		part->chctl.flags[msg->ch_number] |= XPC_CHCTL_OPENREPLY;
		spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

		xpc_wakeup_channel_mgr(part);
		break;
	}
	case XPC_ACTIVATE_MQ_MSG_CHCTL_OPENCOMPLETE_UV: {
		struct xpc_activate_mq_msg_chctl_opencomplete_uv *msg;

		if (!part_setup)
			break;

		msg = container_of(msg_hdr, struct
				xpc_activate_mq_msg_chctl_opencomplete_uv, hdr);
		spin_lock_irqsave(&part->chctl_lock, irq_flags);
		part->chctl.flags[msg->ch_number] |= XPC_CHCTL_OPENCOMPLETE;
		spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

		xpc_wakeup_channel_mgr(part);
	}
	case XPC_ACTIVATE_MQ_MSG_MARK_ENGAGED_UV:
		spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
		part_uv->flags |= XPC_P_ENGAGED_UV;
		spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
		break;

	case XPC_ACTIVATE_MQ_MSG_MARK_DISENGAGED_UV:
		spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
		part_uv->flags &= ~XPC_P_ENGAGED_UV;
		spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
		break;

	default:
		dev_err(xpc_part, "received unknown activate_mq msg type=%d "
			"from partition=%d\n", msg_hdr->type, XPC_PARTID(part));

		/* get hb checker to deactivate from the remote partition */
		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (part_uv->act_state_req == 0)
			xpc_activate_IRQ_rcvd++;
		part_uv->act_state_req = XPC_P_ASR_DEACTIVATE_UV;
		part_uv->reason = xpBadMsgType;
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		(*wakeup_hb_checker)++;
		return;
	}

	if (msg_hdr->rp_ts_jiffies != part->remote_rp_ts_jiffies &&
	    part->remote_rp_ts_jiffies != 0) {
		/*
		 * ??? Does what we do here need to be sensitive to
		 * ??? act_state or remote_act_state?
		 */
		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (part_uv->act_state_req == 0)
			xpc_activate_IRQ_rcvd++;
		part_uv->act_state_req = XPC_P_ASR_REACTIVATE_UV;
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		(*wakeup_hb_checker)++;
	}
}

static irqreturn_t
xpc_handle_activate_IRQ_uv(int irq, void *dev_id)
{
	struct xpc_activate_mq_msghdr_uv *msg_hdr;
	short partid;
	struct xpc_partition *part;
	int wakeup_hb_checker = 0;
	int part_referenced;

	while (1) {
		msg_hdr = gru_get_next_message(xpc_activate_mq_uv->gru_mq_desc);
		if (msg_hdr == NULL)
			break;

		partid = msg_hdr->partid;
		if (partid < 0 || partid >= XP_MAX_NPARTITIONS_UV) {
			dev_err(xpc_part, "xpc_handle_activate_IRQ_uv() "
				"received invalid partid=0x%x in message\n",
				partid);
		} else {
			part = &xpc_partitions[partid];

			part_referenced = xpc_part_ref(part);
			xpc_handle_activate_mq_msg_uv(part, msg_hdr,
						      part_referenced,
						      &wakeup_hb_checker);
			if (part_referenced)
				xpc_part_deref(part);
		}

		gru_free_message(xpc_activate_mq_uv->gru_mq_desc, msg_hdr);
	}

	if (wakeup_hb_checker)
		wake_up_interruptible(&xpc_activate_IRQ_wq);

	return IRQ_HANDLED;
}

static enum xp_retval
xpc_cache_remote_gru_mq_desc_uv(struct gru_message_queue_desc *gru_mq_desc,
				unsigned long gru_mq_desc_gpa)
{
	enum xp_retval ret;

	ret = xp_remote_memcpy(uv_gpa(gru_mq_desc), gru_mq_desc_gpa,
			       sizeof(struct gru_message_queue_desc));
	if (ret == xpSuccess)
		gru_mq_desc->mq = NULL;

	return ret;
}

static enum xp_retval
xpc_send_activate_IRQ_uv(struct xpc_partition *part, void *msg, size_t msg_size,
			 int msg_type)
{
	struct xpc_activate_mq_msghdr_uv *msg_hdr = msg;
	struct xpc_partition_uv *part_uv = &part->sn.uv;
	struct gru_message_queue_desc *gru_mq_desc;
	unsigned long irq_flags;
	enum xp_retval ret;

	DBUG_ON(msg_size > XPC_ACTIVATE_MSG_SIZE_UV);

	msg_hdr->type = msg_type;
	msg_hdr->partid = xp_partition_id;
	msg_hdr->act_state = part->act_state;
	msg_hdr->rp_ts_jiffies = xpc_rsvd_page->ts_jiffies;

	mutex_lock(&part_uv->cached_activate_gru_mq_desc_mutex);
again:
	if (!(part_uv->flags & XPC_P_CACHED_ACTIVATE_GRU_MQ_DESC_UV)) {
		gru_mq_desc = part_uv->cached_activate_gru_mq_desc;
		if (gru_mq_desc == NULL) {
			gru_mq_desc = kmalloc(sizeof(struct
					      gru_message_queue_desc),
					      GFP_KERNEL);
			if (gru_mq_desc == NULL) {
				ret = xpNoMemory;
				goto done;
			}
			part_uv->cached_activate_gru_mq_desc = gru_mq_desc;
		}

		ret = xpc_cache_remote_gru_mq_desc_uv(gru_mq_desc,
						      part_uv->
						      activate_gru_mq_desc_gpa);
		if (ret != xpSuccess)
			goto done;

		spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
		part_uv->flags |= XPC_P_CACHED_ACTIVATE_GRU_MQ_DESC_UV;
		spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
	}

	/* ??? Is holding a spin_lock (ch->lock) during this call a bad idea? */
	ret = xpc_send_gru_msg(part_uv->cached_activate_gru_mq_desc, msg,
			       msg_size);
	if (ret != xpSuccess) {
		smp_rmb();	/* ensure a fresh copy of part_uv->flags */
		if (!(part_uv->flags & XPC_P_CACHED_ACTIVATE_GRU_MQ_DESC_UV))
			goto again;
	}
done:
	mutex_unlock(&part_uv->cached_activate_gru_mq_desc_mutex);
	return ret;
}

static void
xpc_send_activate_IRQ_part_uv(struct xpc_partition *part, void *msg,
			      size_t msg_size, int msg_type)
{
	enum xp_retval ret;

	ret = xpc_send_activate_IRQ_uv(part, msg, msg_size, msg_type);
	if (unlikely(ret != xpSuccess))
		XPC_DEACTIVATE_PARTITION(part, ret);
}

static void
xpc_send_activate_IRQ_ch_uv(struct xpc_channel *ch, unsigned long *irq_flags,
			 void *msg, size_t msg_size, int msg_type)
{
	struct xpc_partition *part = &xpc_partitions[ch->partid];
	enum xp_retval ret;

	ret = xpc_send_activate_IRQ_uv(part, msg, msg_size, msg_type);
	if (unlikely(ret != xpSuccess)) {
		if (irq_flags != NULL)
			spin_unlock_irqrestore(&ch->lock, *irq_flags);

		XPC_DEACTIVATE_PARTITION(part, ret);

		if (irq_flags != NULL)
			spin_lock_irqsave(&ch->lock, *irq_flags);
	}
}

static void
xpc_send_local_activate_IRQ_uv(struct xpc_partition *part, int act_state_req)
{
	unsigned long irq_flags;
	struct xpc_partition_uv *part_uv = &part->sn.uv;

	/*
	 * !!! Make our side think that the remote partition sent an activate
	 * !!! mq message our way by doing what the activate IRQ handler would
	 * !!! do had one really been sent.
	 */

	spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
	if (part_uv->act_state_req == 0)
		xpc_activate_IRQ_rcvd++;
	part_uv->act_state_req = act_state_req;
	spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

	wake_up_interruptible(&xpc_activate_IRQ_wq);
}

static enum xp_retval
xpc_get_partition_rsvd_page_pa_uv(void *buf, u64 *cookie, unsigned long *rp_pa,
				  size_t *len)
{
	s64 status;
	enum xp_retval ret;

#if defined CONFIG_X86_64
	status = uv_bios_reserved_page_pa((u64)buf, cookie, (u64 *)rp_pa,
					  (u64 *)len);
	if (status == BIOS_STATUS_SUCCESS)
		ret = xpSuccess;
	else if (status == BIOS_STATUS_MORE_PASSES)
		ret = xpNeedMoreInfo;
	else
		ret = xpBiosError;

#elif defined CONFIG_IA64_GENERIC || defined CONFIG_IA64_SGI_UV
	status = sn_partition_reserved_page_pa((u64)buf, cookie, rp_pa, len);
	if (status == SALRET_OK)
		ret = xpSuccess;
	else if (status == SALRET_MORE_PASSES)
		ret = xpNeedMoreInfo;
	else
		ret = xpSalError;

#else
	#error not a supported configuration
#endif

	return ret;
}

static int
xpc_setup_rsvd_page_uv(struct xpc_rsvd_page *rp)
{
	xpc_heartbeat_uv =
	    &xpc_partitions[sn_partition_id].sn.uv.cached_heartbeat;
	rp->sn.uv.heartbeat_gpa = uv_gpa(xpc_heartbeat_uv);
	rp->sn.uv.activate_gru_mq_desc_gpa =
	    uv_gpa(xpc_activate_mq_uv->gru_mq_desc);
	return 0;
}

static void
xpc_allow_hb_uv(short partid)
{
}

static void
xpc_disallow_hb_uv(short partid)
{
}

static void
xpc_disallow_all_hbs_uv(void)
{
}

static void
xpc_increment_heartbeat_uv(void)
{
	xpc_heartbeat_uv->value++;
}

static void
xpc_offline_heartbeat_uv(void)
{
	xpc_increment_heartbeat_uv();
	xpc_heartbeat_uv->offline = 1;
}

static void
xpc_online_heartbeat_uv(void)
{
	xpc_increment_heartbeat_uv();
	xpc_heartbeat_uv->offline = 0;
}

static void
xpc_heartbeat_init_uv(void)
{
	xpc_heartbeat_uv->value = 1;
	xpc_heartbeat_uv->offline = 0;
}

static void
xpc_heartbeat_exit_uv(void)
{
	xpc_offline_heartbeat_uv();
}

static enum xp_retval
xpc_get_remote_heartbeat_uv(struct xpc_partition *part)
{
	struct xpc_partition_uv *part_uv = &part->sn.uv;
	enum xp_retval ret;

	ret = xp_remote_memcpy(uv_gpa(&part_uv->cached_heartbeat),
			       part_uv->heartbeat_gpa,
			       sizeof(struct xpc_heartbeat_uv));
	if (ret != xpSuccess)
		return ret;

	if (part_uv->cached_heartbeat.value == part->last_heartbeat &&
	    !part_uv->cached_heartbeat.offline) {

		ret = xpNoHeartbeat;
	} else {
		part->last_heartbeat = part_uv->cached_heartbeat.value;
	}
	return ret;
}

static void
xpc_request_partition_activation_uv(struct xpc_rsvd_page *remote_rp,
				    unsigned long remote_rp_gpa, int nasid)
{
	short partid = remote_rp->SAL_partid;
	struct xpc_partition *part = &xpc_partitions[partid];
	struct xpc_activate_mq_msg_activate_req_uv msg;

	part->remote_rp_pa = remote_rp_gpa; /* !!! _pa here is really _gpa */
	part->remote_rp_ts_jiffies = remote_rp->ts_jiffies;
	part->sn.uv.heartbeat_gpa = remote_rp->sn.uv.heartbeat_gpa;
	part->sn.uv.activate_gru_mq_desc_gpa =
	    remote_rp->sn.uv.activate_gru_mq_desc_gpa;

	/*
	 * ??? Is it a good idea to make this conditional on what is
	 * ??? potentially stale state information?
	 */
	if (part->sn.uv.remote_act_state == XPC_P_AS_INACTIVE) {
		msg.rp_gpa = uv_gpa(xpc_rsvd_page);
		msg.heartbeat_gpa = xpc_rsvd_page->sn.uv.heartbeat_gpa;
		msg.activate_gru_mq_desc_gpa =
		    xpc_rsvd_page->sn.uv.activate_gru_mq_desc_gpa;
		xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
					   XPC_ACTIVATE_MQ_MSG_ACTIVATE_REQ_UV);
	}

	if (part->act_state == XPC_P_AS_INACTIVE)
		xpc_send_local_activate_IRQ_uv(part, XPC_P_ASR_ACTIVATE_UV);
}

static void
xpc_request_partition_reactivation_uv(struct xpc_partition *part)
{
	xpc_send_local_activate_IRQ_uv(part, XPC_P_ASR_ACTIVATE_UV);
}

static void
xpc_request_partition_deactivation_uv(struct xpc_partition *part)
{
	struct xpc_activate_mq_msg_deactivate_req_uv msg;

	/*
	 * ??? Is it a good idea to make this conditional on what is
	 * ??? potentially stale state information?
	 */
	if (part->sn.uv.remote_act_state != XPC_P_AS_DEACTIVATING &&
	    part->sn.uv.remote_act_state != XPC_P_AS_INACTIVE) {

		msg.reason = part->reason;
		xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
					 XPC_ACTIVATE_MQ_MSG_DEACTIVATE_REQ_UV);
	}
}

static void
xpc_cancel_partition_deactivation_request_uv(struct xpc_partition *part)
{
	/* nothing needs to be done */
	return;
}

static void
xpc_init_fifo_uv(struct xpc_fifo_head_uv *head)
{
	head->first = NULL;
	head->last = NULL;
	spin_lock_init(&head->lock);
	head->n_entries = 0;
}

static void *
xpc_get_fifo_entry_uv(struct xpc_fifo_head_uv *head)
{
	unsigned long irq_flags;
	struct xpc_fifo_entry_uv *first;

	spin_lock_irqsave(&head->lock, irq_flags);
	first = head->first;
	if (head->first != NULL) {
		head->first = first->next;
		if (head->first == NULL)
			head->last = NULL;

		head->n_entries--;
		BUG_ON(head->n_entries < 0);

		first->next = NULL;
	}
	spin_unlock_irqrestore(&head->lock, irq_flags);
	return first;
}

static void
xpc_put_fifo_entry_uv(struct xpc_fifo_head_uv *head,
		      struct xpc_fifo_entry_uv *last)
{
	unsigned long irq_flags;

	last->next = NULL;
	spin_lock_irqsave(&head->lock, irq_flags);
	if (head->last != NULL)
		head->last->next = last;
	else
		head->first = last;
	head->last = last;
	head->n_entries++;
	spin_unlock_irqrestore(&head->lock, irq_flags);
}

static int
xpc_n_of_fifo_entries_uv(struct xpc_fifo_head_uv *head)
{
	return head->n_entries;
}

/*
 * Setup the channel structures that are uv specific.
 */
static enum xp_retval
xpc_setup_ch_structures_uv(struct xpc_partition *part)
{
	struct xpc_channel_uv *ch_uv;
	int ch_number;

	for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
		ch_uv = &part->channels[ch_number].sn.uv;

		xpc_init_fifo_uv(&ch_uv->msg_slot_free_list);
		xpc_init_fifo_uv(&ch_uv->recv_msg_list);
	}

	return xpSuccess;
}

/*
 * Teardown the channel structures that are uv specific.
 */
static void
xpc_teardown_ch_structures_uv(struct xpc_partition *part)
{
	/* nothing needs to be done */
	return;
}

static enum xp_retval
xpc_make_first_contact_uv(struct xpc_partition *part)
{
	struct xpc_activate_mq_msg_uv msg;

	/*
	 * We send a sync msg to get the remote partition's remote_act_state
	 * updated to our current act_state which at this point should
	 * be XPC_P_AS_ACTIVATING.
	 */
	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
				      XPC_ACTIVATE_MQ_MSG_SYNC_ACT_STATE_UV);

	while (!((part->sn.uv.remote_act_state == XPC_P_AS_ACTIVATING) ||
		 (part->sn.uv.remote_act_state == XPC_P_AS_ACTIVE))) {

		dev_dbg(xpc_part, "waiting to make first contact with "
			"partition %d\n", XPC_PARTID(part));

		/* wait a 1/4 of a second or so */
		(void)msleep_interruptible(250);

		if (part->act_state == XPC_P_AS_DEACTIVATING)
			return part->reason;
	}

	return xpSuccess;
}

static u64
xpc_get_chctl_all_flags_uv(struct xpc_partition *part)
{
	unsigned long irq_flags;
	union xpc_channel_ctl_flags chctl;

	spin_lock_irqsave(&part->chctl_lock, irq_flags);
	chctl = part->chctl;
	if (chctl.all_flags != 0)
		part->chctl.all_flags = 0;

	spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
	return chctl.all_flags;
}

static enum xp_retval
xpc_allocate_send_msg_slot_uv(struct xpc_channel *ch)
{
	struct xpc_channel_uv *ch_uv = &ch->sn.uv;
	struct xpc_send_msg_slot_uv *msg_slot;
	unsigned long irq_flags;
	int nentries;
	int entry;
	size_t nbytes;

	for (nentries = ch->local_nentries; nentries > 0; nentries--) {
		nbytes = nentries * sizeof(struct xpc_send_msg_slot_uv);
		ch_uv->send_msg_slots = kzalloc(nbytes, GFP_KERNEL);
		if (ch_uv->send_msg_slots == NULL)
			continue;

		for (entry = 0; entry < nentries; entry++) {
			msg_slot = &ch_uv->send_msg_slots[entry];

			msg_slot->msg_slot_number = entry;
			xpc_put_fifo_entry_uv(&ch_uv->msg_slot_free_list,
					      &msg_slot->next);
		}

		spin_lock_irqsave(&ch->lock, irq_flags);
		if (nentries < ch->local_nentries)
			ch->local_nentries = nentries;
		spin_unlock_irqrestore(&ch->lock, irq_flags);
		return xpSuccess;
	}

	return xpNoMemory;
}

static enum xp_retval
xpc_allocate_recv_msg_slot_uv(struct xpc_channel *ch)
{
	struct xpc_channel_uv *ch_uv = &ch->sn.uv;
	struct xpc_notify_mq_msg_uv *msg_slot;
	unsigned long irq_flags;
	int nentries;
	int entry;
	size_t nbytes;

	for (nentries = ch->remote_nentries; nentries > 0; nentries--) {
		nbytes = nentries * ch->entry_size;
		ch_uv->recv_msg_slots = kzalloc(nbytes, GFP_KERNEL);
		if (ch_uv->recv_msg_slots == NULL)
			continue;

		for (entry = 0; entry < nentries; entry++) {
			msg_slot = ch_uv->recv_msg_slots +
			    entry * ch->entry_size;

			msg_slot->hdr.msg_slot_number = entry;
		}

		spin_lock_irqsave(&ch->lock, irq_flags);
		if (nentries < ch->remote_nentries)
			ch->remote_nentries = nentries;
		spin_unlock_irqrestore(&ch->lock, irq_flags);
		return xpSuccess;
	}

	return xpNoMemory;
}

/*
 * Allocate msg_slots associated with the channel.
 */
static enum xp_retval
xpc_setup_msg_structures_uv(struct xpc_channel *ch)
{
	static enum xp_retval ret;
	struct xpc_channel_uv *ch_uv = &ch->sn.uv;

	DBUG_ON(ch->flags & XPC_C_SETUP);

	ch_uv->cached_notify_gru_mq_desc = kmalloc(sizeof(struct
						   gru_message_queue_desc),
						   GFP_KERNEL);
	if (ch_uv->cached_notify_gru_mq_desc == NULL)
		return xpNoMemory;

	ret = xpc_allocate_send_msg_slot_uv(ch);
	if (ret == xpSuccess) {

		ret = xpc_allocate_recv_msg_slot_uv(ch);
		if (ret != xpSuccess) {
			kfree(ch_uv->send_msg_slots);
			xpc_init_fifo_uv(&ch_uv->msg_slot_free_list);
		}
	}
	return ret;
}

/*
 * Free up msg_slots and clear other stuff that were setup for the specified
 * channel.
 */
static void
xpc_teardown_msg_structures_uv(struct xpc_channel *ch)
{
	struct xpc_channel_uv *ch_uv = &ch->sn.uv;

	DBUG_ON(!spin_is_locked(&ch->lock));

	kfree(ch_uv->cached_notify_gru_mq_desc);
	ch_uv->cached_notify_gru_mq_desc = NULL;

	if (ch->flags & XPC_C_SETUP) {
		xpc_init_fifo_uv(&ch_uv->msg_slot_free_list);
		kfree(ch_uv->send_msg_slots);
		xpc_init_fifo_uv(&ch_uv->recv_msg_list);
		kfree(ch_uv->recv_msg_slots);
	}
}

static void
xpc_send_chctl_closerequest_uv(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_activate_mq_msg_chctl_closerequest_uv msg;

	msg.ch_number = ch->number;
	msg.reason = ch->reason;
	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
				    XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREQUEST_UV);
}

static void
xpc_send_chctl_closereply_uv(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_activate_mq_msg_chctl_closereply_uv msg;

	msg.ch_number = ch->number;
	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
				    XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREPLY_UV);
}

static void
xpc_send_chctl_openrequest_uv(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_activate_mq_msg_chctl_openrequest_uv msg;

	msg.ch_number = ch->number;
	msg.entry_size = ch->entry_size;
	msg.local_nentries = ch->local_nentries;
	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
				    XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREQUEST_UV);
}

static void
xpc_send_chctl_openreply_uv(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_activate_mq_msg_chctl_openreply_uv msg;

	msg.ch_number = ch->number;
	msg.local_nentries = ch->local_nentries;
	msg.remote_nentries = ch->remote_nentries;
	msg.notify_gru_mq_desc_gpa = uv_gpa(xpc_notify_mq_uv->gru_mq_desc);
	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
				    XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREPLY_UV);
}

static void
xpc_send_chctl_opencomplete_uv(struct xpc_channel *ch, unsigned long *irq_flags)
{
	struct xpc_activate_mq_msg_chctl_opencomplete_uv msg;

	msg.ch_number = ch->number;
	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
				    XPC_ACTIVATE_MQ_MSG_CHCTL_OPENCOMPLETE_UV);
}

static void
xpc_send_chctl_local_msgrequest_uv(struct xpc_partition *part, int ch_number)
{
	unsigned long irq_flags;

	spin_lock_irqsave(&part->chctl_lock, irq_flags);
	part->chctl.flags[ch_number] |= XPC_CHCTL_MSGREQUEST;
	spin_unlock_irqrestore(&part->chctl_lock, irq_flags);

	xpc_wakeup_channel_mgr(part);
}

static enum xp_retval
xpc_save_remote_msgqueue_pa_uv(struct xpc_channel *ch,
			       unsigned long gru_mq_desc_gpa)
{
	struct xpc_channel_uv *ch_uv = &ch->sn.uv;

	DBUG_ON(ch_uv->cached_notify_gru_mq_desc == NULL);
	return xpc_cache_remote_gru_mq_desc_uv(ch_uv->cached_notify_gru_mq_desc,
					       gru_mq_desc_gpa);
}

static void
xpc_indicate_partition_engaged_uv(struct xpc_partition *part)
{
	struct xpc_activate_mq_msg_uv msg;

	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
				      XPC_ACTIVATE_MQ_MSG_MARK_ENGAGED_UV);
}

static void
xpc_indicate_partition_disengaged_uv(struct xpc_partition *part)
{
	struct xpc_activate_mq_msg_uv msg;

	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
				      XPC_ACTIVATE_MQ_MSG_MARK_DISENGAGED_UV);
}

static void
xpc_assume_partition_disengaged_uv(short partid)
{
	struct xpc_partition_uv *part_uv = &xpc_partitions[partid].sn.uv;
	unsigned long irq_flags;

	spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
	part_uv->flags &= ~XPC_P_ENGAGED_UV;
	spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
}

static int
xpc_partition_engaged_uv(short partid)
{
	return (xpc_partitions[partid].sn.uv.flags & XPC_P_ENGAGED_UV) != 0;
}

static int
xpc_any_partition_engaged_uv(void)
{
	struct xpc_partition_uv *part_uv;
	short partid;

	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
		part_uv = &xpc_partitions[partid].sn.uv;
		if ((part_uv->flags & XPC_P_ENGAGED_UV) != 0)
			return 1;
	}
	return 0;
}

static enum xp_retval
xpc_allocate_msg_slot_uv(struct xpc_channel *ch, u32 flags,
			 struct xpc_send_msg_slot_uv **address_of_msg_slot)
{
	enum xp_retval ret;
	struct xpc_send_msg_slot_uv *msg_slot;
	struct xpc_fifo_entry_uv *entry;

	while (1) {
		entry = xpc_get_fifo_entry_uv(&ch->sn.uv.msg_slot_free_list);
		if (entry != NULL)
			break;

		if (flags & XPC_NOWAIT)
			return xpNoWait;

		ret = xpc_allocate_msg_wait(ch);
		if (ret != xpInterrupted && ret != xpTimeout)
			return ret;
	}

	msg_slot = container_of(entry, struct xpc_send_msg_slot_uv, next);
	*address_of_msg_slot = msg_slot;
	return xpSuccess;
}

static void
xpc_free_msg_slot_uv(struct xpc_channel *ch,
		     struct xpc_send_msg_slot_uv *msg_slot)
{
	xpc_put_fifo_entry_uv(&ch->sn.uv.msg_slot_free_list, &msg_slot->next);

	/* wakeup anyone waiting for a free msg slot */
	if (atomic_read(&ch->n_on_msg_allocate_wq) > 0)
		wake_up(&ch->msg_allocate_wq);
}

static void
xpc_notify_sender_uv(struct xpc_channel *ch,
		     struct xpc_send_msg_slot_uv *msg_slot,
		     enum xp_retval reason)
{
	xpc_notify_func func = msg_slot->func;

	if (func != NULL && cmpxchg(&msg_slot->func, func, NULL) == func) {

		atomic_dec(&ch->n_to_notify);

		dev_dbg(xpc_chan, "msg_slot->func() called, msg_slot=0x%p "
			"msg_slot_number=%d partid=%d channel=%d\n", msg_slot,
			msg_slot->msg_slot_number, ch->partid, ch->number);

		func(reason, ch->partid, ch->number, msg_slot->key);

		dev_dbg(xpc_chan, "msg_slot->func() returned, msg_slot=0x%p "
			"msg_slot_number=%d partid=%d channel=%d\n", msg_slot,
			msg_slot->msg_slot_number, ch->partid, ch->number);
	}
}

static void
xpc_handle_notify_mq_ack_uv(struct xpc_channel *ch,
			    struct xpc_notify_mq_msg_uv *msg)
{
	struct xpc_send_msg_slot_uv *msg_slot;
	int entry = msg->hdr.msg_slot_number % ch->local_nentries;

	msg_slot = &ch->sn.uv.send_msg_slots[entry];

	BUG_ON(msg_slot->msg_slot_number != msg->hdr.msg_slot_number);
	msg_slot->msg_slot_number += ch->local_nentries;

	if (msg_slot->func != NULL)
		xpc_notify_sender_uv(ch, msg_slot, xpMsgDelivered);

	xpc_free_msg_slot_uv(ch, msg_slot);
}

static void
xpc_handle_notify_mq_msg_uv(struct xpc_partition *part,
			    struct xpc_notify_mq_msg_uv *msg)
{
	struct xpc_partition_uv *part_uv = &part->sn.uv;
	struct xpc_channel *ch;
	struct xpc_channel_uv *ch_uv;
	struct xpc_notify_mq_msg_uv *msg_slot;
	unsigned long irq_flags;
	int ch_number = msg->hdr.ch_number;

	if (unlikely(ch_number >= part->nchannels)) {
		dev_err(xpc_part, "xpc_handle_notify_IRQ_uv() received invalid "
			"channel number=0x%x in message from partid=%d\n",
			ch_number, XPC_PARTID(part));

		/* get hb checker to deactivate from the remote partition */
		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
		if (part_uv->act_state_req == 0)
			xpc_activate_IRQ_rcvd++;
		part_uv->act_state_req = XPC_P_ASR_DEACTIVATE_UV;
		part_uv->reason = xpBadChannelNumber;
		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);

		wake_up_interruptible(&xpc_activate_IRQ_wq);
		return;
	}

	ch = &part->channels[ch_number];
	xpc_msgqueue_ref(ch);

	if (!(ch->flags & XPC_C_CONNECTED)) {
		xpc_msgqueue_deref(ch);
		return;
	}

	/* see if we're really dealing with an ACK for a previously sent msg */
	if (msg->hdr.size == 0) {
		xpc_handle_notify_mq_ack_uv(ch, msg);
		xpc_msgqueue_deref(ch);
		return;
	}

	/* we're dealing with a normal message sent via the notify_mq */
	ch_uv = &ch->sn.uv;

	msg_slot = ch_uv->recv_msg_slots +
	    (msg->hdr.msg_slot_number % ch->remote_nentries) * ch->entry_size;

	BUG_ON(msg_slot->hdr.size != 0);

	memcpy(msg_slot, msg, msg->hdr.size);

	xpc_put_fifo_entry_uv(&ch_uv->recv_msg_list, &msg_slot->hdr.u.next);

	if (ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) {
		/*
		 * If there is an existing idle kthread get it to deliver
		 * the payload, otherwise we'll have to get the channel mgr
		 * for this partition to create a kthread to do the delivery.
		 */
		if (atomic_read(&ch->kthreads_idle) > 0)
			wake_up_nr(&ch->idle_wq, 1);
		else
			xpc_send_chctl_local_msgrequest_uv(part, ch->number);
	}
	xpc_msgqueue_deref(ch);
}

static irqreturn_t
xpc_handle_notify_IRQ_uv(int irq, void *dev_id)
{
	struct xpc_notify_mq_msg_uv *msg;
	short partid;
	struct xpc_partition *part;

	while ((msg = gru_get_next_message(xpc_notify_mq_uv->gru_mq_desc)) !=
	       NULL) {

		partid = msg->hdr.partid;
		if (partid < 0 || partid >= XP_MAX_NPARTITIONS_UV) {
			dev_err(xpc_part, "xpc_handle_notify_IRQ_uv() received "
				"invalid partid=0x%x in message\n", partid);
		} else {
			part = &xpc_partitions[partid];

			if (xpc_part_ref(part)) {
				xpc_handle_notify_mq_msg_uv(part, msg);
				xpc_part_deref(part);
			}
		}

		gru_free_message(xpc_notify_mq_uv->gru_mq_desc, msg);
	}

	return IRQ_HANDLED;
}

static int
xpc_n_of_deliverable_payloads_uv(struct xpc_channel *ch)
{
	return xpc_n_of_fifo_entries_uv(&ch->sn.uv.recv_msg_list);
}

static void
xpc_process_msg_chctl_flags_uv(struct xpc_partition *part, int ch_number)
{
	struct xpc_channel *ch = &part->channels[ch_number];
	int ndeliverable_payloads;

	xpc_msgqueue_ref(ch);

	ndeliverable_payloads = xpc_n_of_deliverable_payloads_uv(ch);

	if (ndeliverable_payloads > 0 &&
	    (ch->flags & XPC_C_CONNECTED) &&
	    (ch->flags & XPC_C_CONNECTEDCALLOUT_MADE)) {

		xpc_activate_kthreads(ch, ndeliverable_payloads);
	}

	xpc_msgqueue_deref(ch);
}

static enum xp_retval
xpc_send_payload_uv(struct xpc_channel *ch, u32 flags, void *payload,
		    u16 payload_size, u8 notify_type, xpc_notify_func func,
		    void *key)
{
	enum xp_retval ret = xpSuccess;
	struct xpc_send_msg_slot_uv *msg_slot = NULL;
	struct xpc_notify_mq_msg_uv *msg;
	u8 msg_buffer[XPC_NOTIFY_MSG_SIZE_UV];
	size_t msg_size;

	DBUG_ON(notify_type != XPC_N_CALL);

	msg_size = sizeof(struct xpc_notify_mq_msghdr_uv) + payload_size;
	if (msg_size > ch->entry_size)
		return xpPayloadTooBig;

	xpc_msgqueue_ref(ch);

	if (ch->flags & XPC_C_DISCONNECTING) {
		ret = ch->reason;
		goto out_1;
	}
	if (!(ch->flags & XPC_C_CONNECTED)) {
		ret = xpNotConnected;
		goto out_1;
	}

	ret = xpc_allocate_msg_slot_uv(ch, flags, &msg_slot);
	if (ret != xpSuccess)
		goto out_1;

	if (func != NULL) {
		atomic_inc(&ch->n_to_notify);

		msg_slot->key = key;
		smp_wmb(); /* a non-NULL func must hit memory after the key */
		msg_slot->func = func;

		if (ch->flags & XPC_C_DISCONNECTING) {
			ret = ch->reason;
			goto out_2;
		}
	}

	msg = (struct xpc_notify_mq_msg_uv *)&msg_buffer;
	msg->hdr.partid = xp_partition_id;
	msg->hdr.ch_number = ch->number;
	msg->hdr.size = msg_size;
	msg->hdr.msg_slot_number = msg_slot->msg_slot_number;
	memcpy(&msg->payload, payload, payload_size);

	ret = xpc_send_gru_msg(ch->sn.uv.cached_notify_gru_mq_desc, msg,
			       msg_size);
	if (ret == xpSuccess)
		goto out_1;

	XPC_DEACTIVATE_PARTITION(&xpc_partitions[ch->partid], ret);
out_2:
	if (func != NULL) {
		/*
		 * Try to NULL the msg_slot's func field. If we fail, then
		 * xpc_notify_senders_of_disconnect_uv() beat us to it, in which
		 * case we need to pretend we succeeded to send the message
		 * since the user will get a callout for the disconnect error
		 * by xpc_notify_senders_of_disconnect_uv(), and to also get an
		 * error returned here will confuse them. Additionally, since
		 * in this case the channel is being disconnected we don't need
		 * to put the the msg_slot back on the free list.
		 */
		if (cmpxchg(&msg_slot->func, func, NULL) != func) {
			ret = xpSuccess;
			goto out_1;
		}

		msg_slot->key = NULL;
		atomic_dec(&ch->n_to_notify);
	}
	xpc_free_msg_slot_uv(ch, msg_slot);
out_1:
	xpc_msgqueue_deref(ch);
	return ret;
}

/*
 * Tell the callers of xpc_send_notify() that the status of their payloads
 * is unknown because the channel is now disconnecting.
 *
 * We don't worry about putting these msg_slots on the free list since the
 * msg_slots themselves are about to be kfree'd.
 */
static void
xpc_notify_senders_of_disconnect_uv(struct xpc_channel *ch)
{
	struct xpc_send_msg_slot_uv *msg_slot;
	int entry;

	DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));

	for (entry = 0; entry < ch->local_nentries; entry++) {

		if (atomic_read(&ch->n_to_notify) == 0)
			break;

		msg_slot = &ch->sn.uv.send_msg_slots[entry];
		if (msg_slot->func != NULL)
			xpc_notify_sender_uv(ch, msg_slot, ch->reason);
	}
}

/*
 * Get the next deliverable message's payload.
 */
static void *
xpc_get_deliverable_payload_uv(struct xpc_channel *ch)
{
	struct xpc_fifo_entry_uv *entry;
	struct xpc_notify_mq_msg_uv *msg;
	void *payload = NULL;

	if (!(ch->flags & XPC_C_DISCONNECTING)) {
		entry = xpc_get_fifo_entry_uv(&ch->sn.uv.recv_msg_list);
		if (entry != NULL) {
			msg = container_of(entry, struct xpc_notify_mq_msg_uv,
					   hdr.u.next);
			payload = &msg->payload;
		}
	}
	return payload;
}

static void
xpc_received_payload_uv(struct xpc_channel *ch, void *payload)
{
	struct xpc_notify_mq_msg_uv *msg;
	enum xp_retval ret;

	msg = container_of(payload, struct xpc_notify_mq_msg_uv, payload);

	/* return an ACK to the sender of this message */

	msg->hdr.partid = xp_partition_id;
	msg->hdr.size = 0;	/* size of zero indicates this is an ACK */

	ret = xpc_send_gru_msg(ch->sn.uv.cached_notify_gru_mq_desc, msg,
			       sizeof(struct xpc_notify_mq_msghdr_uv));
	if (ret != xpSuccess)
		XPC_DEACTIVATE_PARTITION(&xpc_partitions[ch->partid], ret);
}

static struct xpc_arch_operations xpc_arch_ops_uv = {
	.setup_partitions = xpc_setup_partitions_uv,
	.teardown_partitions = xpc_teardown_partitions_uv,
	.process_activate_IRQ_rcvd = xpc_process_activate_IRQ_rcvd_uv,
	.get_partition_rsvd_page_pa = xpc_get_partition_rsvd_page_pa_uv,
	.setup_rsvd_page = xpc_setup_rsvd_page_uv,

	.allow_hb = xpc_allow_hb_uv,
	.disallow_hb = xpc_disallow_hb_uv,
	.disallow_all_hbs = xpc_disallow_all_hbs_uv,
	.increment_heartbeat = xpc_increment_heartbeat_uv,
	.offline_heartbeat = xpc_offline_heartbeat_uv,
	.online_heartbeat = xpc_online_heartbeat_uv,
	.heartbeat_init = xpc_heartbeat_init_uv,
	.heartbeat_exit = xpc_heartbeat_exit_uv,
	.get_remote_heartbeat = xpc_get_remote_heartbeat_uv,

	.request_partition_activation =
		xpc_request_partition_activation_uv,
	.request_partition_reactivation =
		xpc_request_partition_reactivation_uv,
	.request_partition_deactivation =
		xpc_request_partition_deactivation_uv,
	.cancel_partition_deactivation_request =
		xpc_cancel_partition_deactivation_request_uv,

	.setup_ch_structures = xpc_setup_ch_structures_uv,
	.teardown_ch_structures = xpc_teardown_ch_structures_uv,

	.make_first_contact = xpc_make_first_contact_uv,

	.get_chctl_all_flags = xpc_get_chctl_all_flags_uv,
	.send_chctl_closerequest = xpc_send_chctl_closerequest_uv,
	.send_chctl_closereply = xpc_send_chctl_closereply_uv,
	.send_chctl_openrequest = xpc_send_chctl_openrequest_uv,
	.send_chctl_openreply = xpc_send_chctl_openreply_uv,
	.send_chctl_opencomplete = xpc_send_chctl_opencomplete_uv,
	.process_msg_chctl_flags = xpc_process_msg_chctl_flags_uv,

	.save_remote_msgqueue_pa = xpc_save_remote_msgqueue_pa_uv,

	.setup_msg_structures = xpc_setup_msg_structures_uv,
	.teardown_msg_structures = xpc_teardown_msg_structures_uv,

	.indicate_partition_engaged = xpc_indicate_partition_engaged_uv,
	.indicate_partition_disengaged = xpc_indicate_partition_disengaged_uv,
	.assume_partition_disengaged = xpc_assume_partition_disengaged_uv,
	.partition_engaged = xpc_partition_engaged_uv,
	.any_partition_engaged = xpc_any_partition_engaged_uv,

	.n_of_deliverable_payloads = xpc_n_of_deliverable_payloads_uv,
	.send_payload = xpc_send_payload_uv,
	.get_deliverable_payload = xpc_get_deliverable_payload_uv,
	.received_payload = xpc_received_payload_uv,
	.notify_senders_of_disconnect = xpc_notify_senders_of_disconnect_uv,
};

static int
xpc_init_mq_node(int nid)
{
	int cpu;

	get_online_cpus();

	for_each_cpu(cpu, cpumask_of_node(nid)) {
		xpc_activate_mq_uv =
			xpc_create_gru_mq_uv(XPC_ACTIVATE_MQ_SIZE_UV, nid,
					     XPC_ACTIVATE_IRQ_NAME,
					     xpc_handle_activate_IRQ_uv);
		if (!IS_ERR(xpc_activate_mq_uv))
			break;
	}
	if (IS_ERR(xpc_activate_mq_uv)) {
		put_online_cpus();
		return PTR_ERR(xpc_activate_mq_uv);
	}

	for_each_cpu(cpu, cpumask_of_node(nid)) {
		xpc_notify_mq_uv =
			xpc_create_gru_mq_uv(XPC_NOTIFY_MQ_SIZE_UV, nid,
					     XPC_NOTIFY_IRQ_NAME,
					     xpc_handle_notify_IRQ_uv);
		if (!IS_ERR(xpc_notify_mq_uv))
			break;
	}
	if (IS_ERR(xpc_notify_mq_uv)) {
		xpc_destroy_gru_mq_uv(xpc_activate_mq_uv);
		put_online_cpus();
		return PTR_ERR(xpc_notify_mq_uv);
	}

	put_online_cpus();
	return 0;
}

int
xpc_init_uv(void)
{
	int nid;
	int ret = 0;

	xpc_arch_ops = xpc_arch_ops_uv;

	if (sizeof(struct xpc_notify_mq_msghdr_uv) > XPC_MSG_HDR_MAX_SIZE) {
		dev_err(xpc_part, "xpc_notify_mq_msghdr_uv is larger than %d\n",
			XPC_MSG_HDR_MAX_SIZE);
		return -E2BIG;
	}

	if (xpc_mq_node < 0)
		for_each_online_node(nid) {
			ret = xpc_init_mq_node(nid);

			if (!ret)
				break;
		}
	else
		ret = xpc_init_mq_node(xpc_mq_node);

	if (ret < 0)
		dev_err(xpc_part, "xpc_init_mq_node() returned error=%d\n",
			-ret);

	return ret;
}

void
xpc_exit_uv(void)
{
	xpc_destroy_gru_mq_uv(xpc_notify_mq_uv);
	xpc_destroy_gru_mq_uv(xpc_activate_mq_uv);
}

module_param(xpc_mq_node, int, 0);
MODULE_PARM_DESC(xpc_mq_node, "Node number on which to allocate message queues.");