/****************************************************************************
 * drivers/sensors/sensor_rpmsg.c
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.  The
 * ASF licenses this file to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the
 * License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 *
 ****************************************************************************/

/****************************************************************************
 * Included Files
 ****************************************************************************/

#include <nuttx/config.h>

#include <fcntl.h>
#include <debug.h>
#include <libgen.h>

#include <nuttx/nuttx.h>
#include <nuttx/list.h>
#include <nuttx/kmalloc.h>
#include <nuttx/mutex.h>
#include <nuttx/rwsem.h>
#include <nuttx/sensors/sensor.h>
#include <nuttx/rpmsg/rpmsg.h>

/****************************************************************************
 * Pre-processor Definitions
 ****************************************************************************/

#define SENSOR_RPMSG_EPT_NAME      "rpmsg-sensor"
#define SENSOR_RPMSG_ADVERTISE     0
#define SENSOR_RPMSG_ADVERTISE_ACK 1
#define SENSOR_RPMSG_UNADVERTISE   2
#define SENSOR_RPMSG_SUBSCRIBE     3
#define SENSOR_RPMSG_SUBSCRIBE_ACK 4
#define SENSOR_RPMSG_UNSUBSCRIBE   5
#define SENSOR_RPMSG_PUBLISH       6
#define SENSOR_RPMSG_IOCTL         7
#define SENSOR_RPMSG_IOCTL_ACK     8
#define SENSOR_RPMSG_IOCTL_TIMEOUT 1

#define SENSOR_RPMSG_FUNCTION(name, cmd, arg1, arg2, size, wait, type) \
static int sensor_rpmsg_##name(FAR struct sensor_lowerhalf_s *lower, \
                               FAR struct file *filep, \
                               type arg1) \
{ \
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv; \
  FAR struct sensor_lowerhalf_s *drv = dev->drv; \
  int ret; \
\
  if (drv->ops->name) \
    { \
      return drv->ops->name(drv, filep, arg2); \
    } \
  else if (!(filep->f_oflags & SENSOR_REMOTE)) \
    { \
      ret = sensor_rpmsg_ioctl(dev, cmd, arg1, size, wait); \
      return wait ? ret : 0; \
    } \
  else \
    { \
      return wait ? -ENOTSUP : 0; \
    } \
} \

/****************************************************************************
 * Private Types
 ****************************************************************************/

/* This structure describes the context of sensor rpmsg device driver. */

struct sensor_rpmsg_dev_s
{
  struct sensor_lowerhalf_s      lower;
  FAR struct sensor_lowerhalf_s *drv;
  struct list_node               node;
  struct list_node               stublist;
  struct list_node               proxylist;
  sem_t                          proxysem;
  int16_t                        nadvertisers;
  int16_t                        nsubscribers;
  FAR void                      *upper;
  sensor_push_event_t            push_event;
  sensor_notify_interval_t       notify_interval;
  FAR const char                *name;
  char                           path[1];
};

/* This structure describes the context of sensor rpmsg endpoint. */

struct sensor_rpmsg_ept_s
{
  struct list_node               node;
  struct rpmsg_endpoint          ept;
  struct work_s                  work;
  struct work_s                  bound_work;
  rmutex_t                       lock;
  FAR void                      *buffer;
  uint64_t                       expire;
  uint32_t                       space;
  size_t                         written;
};

/* This structure describes the stub info about remote subscribers. */

struct sensor_rpmsg_stub_s
{
  struct list_node               node;
  FAR struct rpmsg_endpoint     *ept;
  uint64_t                       cookie;
  struct file                    file;
  bool                           flushing;
  bool                           wakeup;
};

/* This structure describes the proxy info about remote advertisers. */

struct sensor_rpmsg_proxy_s
{
  struct list_node               node;
  FAR struct rpmsg_endpoint     *ept;
  uint64_t                       cookie;
  bool                           wakeup;
};

/* Remote message structure */

/* This structure describes the message about initiating a remote
 * subscription and remote advertisement.
 */

struct sensor_rpmsg_advsub_s
{
  uint32_t                       command;
  uint32_t                       nbuffer;
  uint64_t                       cookie;
  uint32_t                       persist;
  char                           path[1];
};

/* The structure sensor_rpmsg_cell_s describes a data message,
 * include remote receiver, the length of data and the data payload.
 * The structure sensor_rpmsg_data_s describes a set of data message.
 */

struct sensor_rpmsg_cell_s
{
  uint64_t                       cookie;
  uint32_t                       nbuffer;
  uint32_t                       len;
  char                           data[0];
};

struct sensor_rpmsg_data_s
{
  uint32_t                       command;
  uint32_t                       reserved;
  struct sensor_rpmsg_cell_s     cell[0];
};

/* This structure uses to send ioctl from remote device to physical device,
 * it supports pass pointer to remove device by cookie_xx member.
 */

struct sensor_rpmsg_ioctl_cookie_s
{
  sem_t                          sem;
  FAR void                      *data;
  int                            result;
};

struct sensor_rpmsg_ioctl_s
{
  uint32_t                       command;
  int32_t                        result;
  uint64_t                       cookie;
  uint64_t                       proxy;

  int32_t                        request;
  uint32_t                       arglen;
  char                           arg[];
};

/****************************************************************************
 * Private Function Prototypes
 ****************************************************************************/

static int sensor_rpmsg_open(FAR struct sensor_lowerhalf_s *lower,
                             FAR struct file *filep);
static int sensor_rpmsg_close(FAR struct sensor_lowerhalf_s *lower,
                              FAR struct file *filep);
static int sensor_rpmsg_activate(FAR struct sensor_lowerhalf_s *lower,
                                 FAR struct file *filep,
                                 bool enable);
static int sensor_rpmsg_set_interval(FAR struct sensor_lowerhalf_s *lower,
                                     FAR struct file *filep,
                                     FAR uint32_t *period_us);
static int sensor_rpmsg_batch(FAR struct sensor_lowerhalf_s *lower,
                              FAR struct file *filep,
                              FAR uint32_t *latency_us);
static int sensor_rpmsg_flush(FAR struct sensor_lowerhalf_s *lower,
                              FAR struct file *filep);
static int sensor_rpmsg_selftest(FAR struct sensor_lowerhalf_s *lower,
                                 FAR struct file *filep,
                                 unsigned long arg);
static int sensor_rpmsg_set_calibvalue(FAR struct sensor_lowerhalf_s *lower,
                                       FAR struct file *filep,
                                       unsigned long arg);
static int sensor_rpmsg_calibrate(FAR struct sensor_lowerhalf_s *lower,
                                  FAR struct file *filep,
                                  unsigned long arg);
static int sensor_rpmsg_get_info(FAR struct sensor_lowerhalf_s *lower,
                                 FAR struct file *filep,
                                 FAR struct sensor_device_info_s *info);
static int sensor_rpmsg_set_wakeup(FAR struct sensor_lowerhalf_s *lower,
                                   FAR struct file *filep,
                                   bool wakeup);
static int sensor_rpmsg_control(FAR struct sensor_lowerhalf_s *lower,
                                FAR struct file *filep,
                                int cmd, unsigned long arg);
static int sensor_rpmsg_adv_handler(FAR struct rpmsg_endpoint *ept,
                                    FAR void *data, size_t len,
                                    uint32_t src, FAR void *priv);
static int sensor_rpmsg_advack_handler(FAR struct rpmsg_endpoint *ept,
                                       FAR void *data, size_t len,
                                       uint32_t src, FAR void *priv);
static int sensor_rpmsg_unadv_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv);
static int sensor_rpmsg_sub_handler(FAR struct rpmsg_endpoint *ept,
                                    FAR void *data, size_t len,
                                    uint32_t src, FAR void *priv);
static int sensor_rpmsg_suback_handler(FAR struct rpmsg_endpoint *ept,
                                       FAR void *data, size_t len,
                                       uint32_t src, FAR void *priv);
static int sensor_rpmsg_unsub_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv);
static int sensor_rpmsg_publish_handler(FAR struct rpmsg_endpoint *ept,
                                        FAR void *data, size_t len,
                                        uint32_t src, FAR void *priv);
static int sensor_rpmsg_ioctl_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv);
static int sensor_rpmsg_ioctlack_handler(FAR struct rpmsg_endpoint *ept,
                                         FAR void *data, size_t len,
                                         uint32_t src, FAR void *priv);

/****************************************************************************
 * Private Data
 ****************************************************************************/

static const struct sensor_ops_s g_sensor_rpmsg_ops =
{
  .open           = sensor_rpmsg_open,
  .close          = sensor_rpmsg_close,
  .activate       = sensor_rpmsg_activate,
  .set_interval   = sensor_rpmsg_set_interval,
  .batch          = sensor_rpmsg_batch,
  .flush          = sensor_rpmsg_flush,
  .selftest       = sensor_rpmsg_selftest,
  .set_calibvalue = sensor_rpmsg_set_calibvalue,
  .calibrate      = sensor_rpmsg_calibrate,
  .get_info       = sensor_rpmsg_get_info,
  .set_wakeup     = sensor_rpmsg_set_wakeup,
  .control        = sensor_rpmsg_control
};

static const rpmsg_ept_cb g_sensor_rpmsg_handler[] =
{
  [SENSOR_RPMSG_ADVERTISE]     = sensor_rpmsg_adv_handler,
  [SENSOR_RPMSG_ADVERTISE_ACK] = sensor_rpmsg_advack_handler,
  [SENSOR_RPMSG_UNADVERTISE]   = sensor_rpmsg_unadv_handler,
  [SENSOR_RPMSG_SUBSCRIBE]     = sensor_rpmsg_sub_handler,
  [SENSOR_RPMSG_SUBSCRIBE_ACK] = sensor_rpmsg_suback_handler,
  [SENSOR_RPMSG_UNSUBSCRIBE]   = sensor_rpmsg_unsub_handler,
  [SENSOR_RPMSG_PUBLISH]       = sensor_rpmsg_publish_handler,
  [SENSOR_RPMSG_IOCTL]         = sensor_rpmsg_ioctl_handler,
  [SENSOR_RPMSG_IOCTL_ACK]     = sensor_rpmsg_ioctlack_handler,
};

static struct list_node g_devlist = LIST_INITIAL_VALUE(g_devlist);
static struct list_node g_eptlist = LIST_INITIAL_VALUE(g_eptlist);
static rw_semaphore_t g_ept_lock = RWSEM_INITIALIZER;
static rw_semaphore_t g_dev_lock = RWSEM_INITIALIZER;

/****************************************************************************
 * Private Functions
 ****************************************************************************/

static void sensor_rpmsg_lock(FAR struct sensor_rpmsg_dev_s *dev)
{
  dev->lower.sensor_lock(dev->upper);
}

static void sensor_rpmsg_unlock(FAR struct sensor_rpmsg_dev_s *dev)
{
  dev->lower.sensor_unlock(dev->upper);
}

static void sensor_rpmsg_advsub_one(FAR struct sensor_rpmsg_dev_s *dev,
                                    FAR struct rpmsg_endpoint *ept,
                                    int command)
{
  FAR struct sensor_rpmsg_advsub_s *msg;
  uint32_t space;
  int len = strlen(dev->path) + 1;
  int ret;

  msg = rpmsg_get_tx_payload_buffer(ept, &space, true);
  if (!msg)
    {
      snerr("ERROR: advsub:%d get buffer failed:%s, %s\n",
            command, dev->path, rpmsg_get_cpuname(ept->rdev));
      return;
    }

  msg->command = command;
  msg->cookie  = (uint64_t)(uintptr_t)dev;
  msg->nbuffer = dev->lower.nbuffer;
  msg->persist = dev->lower.persist;
  memcpy(msg->path, dev->path, len);
  ret = rpmsg_send_nocopy(ept, msg, sizeof(*msg) + len);
  if (ret < 0)
    {
      rpmsg_release_tx_buffer(ept, msg);
      snerr("ERROR: advsub:%d rpmsg send failed:%s, %d, %s\n",
            command, dev->path, ret, rpmsg_get_cpuname(ept->rdev));
    }
}

static void sensor_rpmsg_advsub(FAR struct sensor_rpmsg_dev_s *dev,
                                int command)
{
  FAR struct sensor_rpmsg_ept_s *sre;

  /* Broadcast advertise/subscribe message to all ready ept */

  down_read(&g_ept_lock);
  list_for_every_entry(&g_eptlist, sre, struct sensor_rpmsg_ept_s,
                       node)
    {
      FAR struct sensor_rpmsg_proxy_s *proxy = NULL;
      bool find = false;
      bool adv = false;

      if (command == SENSOR_RPMSG_SUBSCRIBE ||
          command == SENSOR_RPMSG_UNSUBSCRIBE)
        {
          /* If it has been bound before, do not send. */

          list_for_every_entry(&dev->proxylist, proxy,
                               struct sensor_rpmsg_proxy_s, node)
            {
              if (proxy->ept == &sre->ept)
                {
                  find = true;
                  break;
                }
            }
        }
      else
        {
          /* Advertise and unadvertise need to be broadcast. otherwise, data
           * will be lost if the following two conditions occur at the same
           * time.
           *  1.The adv core crashed, the advertiser has a long life cycle.
           *  2.The remote subscriber has a long life cycle, and the remote
           *    core side has not been restarted
           */

          adv = true;
        }

      /* There are several scenarios that require broadcasting:
       * 1. If the proxy corresponding to the EPT (Endpoint) did not
       *    exist previously, this constitutes the first broadcast.
       * 2. If the proxy previously had a wakeup attribute, it should
       *    be broadcast every time.
       * 3. If the proxy does not have the wakeup attribute and the
       *    target core is in the running state, we should still broadcast
       *    it.
       *
       * In summary, Let's avoid broadcasting non-wakeup sensors when the
       * target core is in a sleep state to prevent unnecessary wakeups.
       */

      if (adv || rpmsg_is_running(sre->ept.rdev) || (find && proxy->wakeup))
        {
          sensor_rpmsg_advsub_one(dev, &sre->ept, command);
        }
      else
        {
          sninfo("INFO: advsub:%d %s ignore broadcast to %s, "
                 "as %d, %d, %d\n",
                 command, dev->path, rpmsg_get_cpuname(sre->ept.rdev),
                 find, proxy ? proxy->wakeup : 0,
                 rpmsg_is_running(sre->ept.rdev));
        }
    }

  up_read(&g_ept_lock);
}

static int sensor_rpmsg_ioctl(FAR struct sensor_rpmsg_dev_s *dev,
                              int cmd, unsigned long arg, size_t len,
                              bool wait)
{
  struct sensor_rpmsg_ioctl_cookie_s cookie;
  FAR struct sensor_rpmsg_proxy_s *proxy;
  FAR struct sensor_rpmsg_proxy_s *ptmp;
  FAR struct sensor_rpmsg_ioctl_s *msg;
  FAR struct rpmsg_endpoint *ept;
  uint64_t pcookie;
  size_t send_len;
  uint32_t space;
  int ret = -ENOTTY;
  bool empty;

  /* All control is always send to own proxy(remote advertisers),
   * if device doesn't have proxy, need to wait until proxy is created.
   */

  sensor_rpmsg_lock(dev);
  empty = list_is_empty(&dev->proxylist);
  sensor_rpmsg_unlock(dev);

  if (wait)
    {
      if (empty)
        {
          ret = nxsem_tickwait_uninterruptible(
              &dev->proxysem, SEC2TICK(SENSOR_RPMSG_IOCTL_TIMEOUT));
          if (ret < 0)
            {
              snerr("ERROR: wait proxy create failed:%s, cmd:%d\n",
                    dev->path, cmd);
              return ret;
            }

          nxsem_post(&dev->proxysem);
        }

      cookie.data   = (FAR void *)(uintptr_t)arg;
      cookie.result = -ENXIO;
      nxsem_init(&cookie.sem, 0, 0);
    }

  send_len = len > 0 ? len : sizeof(uint64_t);
  sensor_rpmsg_lock(dev);
  list_for_every_entry_safe(&dev->proxylist, proxy, ptmp,
                            struct sensor_rpmsg_proxy_s, node)
    {
      /* Check whether the proxy has been set to the wakeup state
       * before. If it has, record it.
       */

      if (cmd == SNIOC_SET_WAKEUP)
        {
          proxy->wakeup = arg;
        }

      ept = proxy->ept;
      pcookie = proxy->cookie;
      msg = rpmsg_get_tx_payload_buffer(ept, &space, true);
      if (!msg)
        {
          ret = -ENOMEM;
          snerr("ERROR: ioctl get buffer failed:%s, %s\n",
                dev->path, rpmsg_get_cpuname(ept->rdev));
          break;
        }

      msg->command = SENSOR_RPMSG_IOCTL;
      msg->cookie  = wait ? (uint64_t)(uintptr_t)&cookie : 0;
      msg->proxy   = pcookie;
      msg->request = cmd;
      msg->arglen  = len;
      if (len > 0)
        {
          memcpy(msg->arg, (FAR void *)(uintptr_t)arg, len);
        }
      else
        {
          *(FAR uint64_t *)msg->arg = arg;
        }

      ret = rpmsg_send_nocopy(ept, msg, sizeof(*msg) + send_len);
      if (ret < 0)
        {
          rpmsg_release_tx_buffer(ept, msg);
          snerr("ERROR: ioctl rpmsg send failed:%s, %d, %s\n",
                dev->path, ret, rpmsg_get_cpuname(ept->rdev));
          break;
        }

      if (!wait)
        {
          continue;
        }

      sensor_rpmsg_unlock(dev);
      ret = rpmsg_wait(ept, &cookie.sem);
      sensor_rpmsg_lock(dev);
      if (ret < 0)
        {
          snerr("ERROR: ioctl rpmsg wait failed:%s, %d, %s\n",
                dev->path, ret, rpmsg_get_cpuname(ept->rdev));
          break;
        }

      ret = cookie.result;
      if (ret < 0 && ret != -ENOTTY)
        {
          break;
        }
    }

  sensor_rpmsg_unlock(dev);
  if (wait)
    {
      nxsem_destroy(&cookie.sem);
    }

  return ret;
}

static FAR struct sensor_rpmsg_proxy_s *
sensor_rpmsg_alloc_proxy(FAR struct sensor_rpmsg_dev_s *dev,
                         FAR struct rpmsg_endpoint *ept,
                         FAR struct sensor_rpmsg_advsub_s *msg,
                         FAR struct sensor_state_s *state)
{
  FAR struct sensor_rpmsg_proxy_s *proxy;
  struct file file;
  int ret;

  sensor_rpmsg_lock(dev);
  list_for_every_entry(&dev->proxylist, proxy,
                       struct sensor_rpmsg_proxy_s, node)
    {
      if (proxy->ept == ept && proxy->cookie == msg->cookie)
        {
          sensor_rpmsg_unlock(dev);
          return proxy;
        }
    }

  sensor_rpmsg_unlock(dev);

  /* Create new proxy to represent a remote advertiser */

  proxy = kmm_zalloc(sizeof(*proxy));
  if (!proxy)
    {
      return NULL;
    }

  proxy->ept = ept;
  proxy->cookie = msg->cookie;
  ret = file_open(&file, dev->path, SENSOR_REMOTE | O_CLOEXEC);
  if (ret < 0)
    {
      kmm_free(proxy);
      return NULL;
    }

  file_ioctl(&file, SNIOC_SET_BUFFER_NUMBER, msg->nbuffer);
  file_ioctl(&file, SNIOC_GET_STATE, state);
  file_close(&file);

  sensor_rpmsg_lock(dev);
  if (msg->persist)
    {
      dev->drv->persist = true;
      dev->lower.persist = true;
    }

  list_add_tail(&dev->proxylist, &proxy->node);
  nxsem_post(&dev->proxysem);
  sensor_rpmsg_unlock(dev);
  sminfo(dev->name, "create proxy:%p", proxy);
  return proxy;
}

static FAR struct sensor_rpmsg_stub_s *
sensor_rpmsg_push_event_persist(FAR struct sensor_rpmsg_dev_s *dev,
                                FAR struct rpmsg_endpoint *ept,
                                FAR struct sensor_rpmsg_stub_s *stub,
                                uint64_t cookie)
{
  FAR struct sensor_rpmsg_cell_s *cell;
  FAR struct sensor_rpmsg_stub_s *temp;
  FAR struct sensor_rpmsg_data_s *msg;
  FAR struct sensor_rpmsg_ept_s *sre;
  bool find = false;
  uint32_t space;
  int ret;

  sre = container_of(stub->ept, struct sensor_rpmsg_ept_s, ept);
  msg = rpmsg_get_tx_payload_buffer(&sre->ept, &space, true);
  if (!msg)
    {
      snerr("ERROR: push event persist get buffer failed:%s\n",
            rpmsg_get_cpuname(sre->ept.rdev));
      return NULL;
    }

  /* Check if the stub still exists.
   * rpmsg involves recursive calls, which may result in the sub_handler
   * not completing its execution, the rpmsg recursively switches to
   * the unsub_handler and releases the stub.
   */

  list_for_every_entry(&dev->stublist, temp,
                       struct sensor_rpmsg_stub_s, node)
    {
      /* Check if this stub matches the given ept and cookie. */

      if (temp->ept == ept && temp->cookie == cookie)
        {
          find = true;
          break;
        }
    }

  /* The current position requires determining whether the stub
   * node exists. there are two possibilities.
   * 1.The node exists, and it's still the address originally allocated
   *   by malloc.
   * 2.The node exists, but the address has changed due to reallocation.
   *   the stub address needs to be updated. Return to the new address
   *   of the stub
   * 3.The node does not exist anymore，return directly.
   */

  if (!find)
    {
      rpmsg_release_tx_buffer(&sre->ept, msg);
      return NULL;
    }
  else if (stub != temp)
    {
      stub = temp;
      snwarn("WARN: push event persist stub changed:%p -> %p\n", stub, temp);
    }

  msg->command = SENSOR_RPMSG_PUBLISH;
  cell = (FAR struct sensor_rpmsg_cell_s *)(msg + 1);
  ret = file_read(&stub->file, cell->data, space - sizeof(*msg) -
                  sizeof(*cell));
  if (ret > 0)
    {
      cell->len     = ret;
      cell->cookie  = stub->cookie;
      cell->nbuffer = dev->lower.nbuffer;
      rpmsg_send_nocopy(&sre->ept, msg, sizeof(*msg) +
                        ((sizeof(*cell) + ret + 0x7) & ~0x7));
    }
  else
    {
      rpmsg_release_tx_buffer(&sre->ept, msg);
    }

  return stub;
}

static FAR struct sensor_rpmsg_stub_s *
sensor_rpmsg_alloc_stub(FAR struct sensor_rpmsg_dev_s *dev,
                        FAR struct rpmsg_endpoint *ept,
                        uint64_t cookie)
{
  FAR struct sensor_rpmsg_stub_s *stub;
  int ret;

  sensor_rpmsg_lock(dev);
  list_for_every_entry(&dev->stublist, stub,
                       struct sensor_rpmsg_stub_s, node)
    {
      if (stub->ept == ept && stub->cookie == cookie)
        {
          sensor_rpmsg_unlock(dev);
          return stub;
        }
    }

  sensor_rpmsg_unlock(dev);

  /* Create new stub to represent a remote subscribers */

  stub = kmm_zalloc(sizeof(*stub));
  if (!stub)
    {
      return NULL;
    }

  stub->ept = ept;
  stub->cookie = cookie;
  ret = file_open(&stub->file, dev->path,
                  O_RDOK | O_NONBLOCK | O_CLOEXEC | SENSOR_REMOTE);
  if (ret < 0)
    {
      kmm_free(stub);
      return NULL;
    }

  sensor_rpmsg_lock(dev);
  list_add_tail(&dev->stublist, &stub->node);

  if (dev->lower.persist)
    {
      stub = sensor_rpmsg_push_event_persist(dev, ept, stub, cookie);
      sminfo(dev->name, "push event persist status:%p", stub);
    }

  sensor_rpmsg_unlock(dev);
  sminfo(dev->name, "create stub:%p", stub);

  return stub;
}

static void sensor_rpmsg_free_proxy(FAR struct sensor_rpmsg_dev_s *dev,
                                    FAR struct sensor_rpmsg_proxy_s *proxy)
{
  sminfo(dev->name, "free proxy:%p name:%s", proxy, proxy->ept->name);
  list_delete(&proxy->node);
  kmm_free(proxy);
  if (list_is_empty(&dev->proxylist))
    {
      nxsem_reset(&dev->proxysem, 0);
    }
}

static void sensor_rpmsg_free_stub(FAR struct sensor_rpmsg_stub_s *stub)
{
  list_delete(&stub->node);
  file_close(&stub->file);
  kmm_free(stub);
}

static int sensor_rpmsg_open(FAR struct sensor_lowerhalf_s *lower,
                             FAR struct file *filep)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  bool adv = false;
  bool sub = false;
  int ret;

  if (drv->ops->open)
    {
      ret = drv->ops->open(drv, filep);
      if (ret < 0)
        {
          return ret;
        }
    }

  if (filep->f_oflags & SENSOR_REMOTE)
    {
      return 0;
    }

  sensor_rpmsg_lock(dev);
  if (filep->f_oflags & O_WROK)
    {
      if (filep->f_oflags & SENSOR_PERSIST)
        {
          dev->lower.persist = true;
        }

      if (dev->nadvertisers++ <= 0)
        {
          dev->nadvertisers = 1;
          adv = true;
        }
    }

  if (filep->f_oflags & O_RDOK)
    {
      /* Send broadcast subscribed info if dev isn't
       * physical sensor to avoid waking up remote core.
       */

      if (dev->nsubscribers++ == 0 &&
          drv->ops->activate == NULL)
        {
          sub = true;
        }
    }

  sensor_rpmsg_unlock(dev);

  if (adv)
    {
      sensor_rpmsg_advsub(dev, SENSOR_RPMSG_ADVERTISE);
    }

  if (sub)
    {
      sensor_rpmsg_advsub(dev, SENSOR_RPMSG_SUBSCRIBE);
    }

  return 0;
}

static int sensor_rpmsg_close(FAR struct sensor_lowerhalf_s *lower,
                              FAR struct file *filep)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  FAR struct sensor_rpmsg_stub_s *stub;
  FAR struct sensor_rpmsg_stub_s *stmp;
  bool unsub = false;
  int ret = 0;

  if (drv->ops->close)
    {
      ret = drv->ops->close(drv, filep);
    }

  if (filep->f_oflags & SENSOR_REMOTE)
    {
      return ret;
    }

  sensor_rpmsg_lock(dev);
  if (filep->f_oflags & O_WROK)
    {
      if (dev->nadvertisers == 1)
        {
          list_for_every_entry_safe(&dev->stublist, stub, stmp,
                                    struct sensor_rpmsg_stub_s, node)
            {
              sensor_rpmsg_free_stub(stub);
            }
        }

      dev->nadvertisers--;
    }

  if (filep->f_oflags & O_RDOK)
    {
      /* Send broadcast unsubscribed info if dev isn't
       * physical sensor to avoid waking up remote core.
       */

      if (dev->nsubscribers == 1 &&
          drv->ops->activate == NULL)
        {
          unsub = true;
        }

      dev->nsubscribers--;
    }

  sensor_rpmsg_unlock(dev);

  if (unsub)
    {
      sensor_rpmsg_advsub(dev, SENSOR_RPMSG_UNSUBSCRIBE);
    }

  return ret;
}

static int sensor_rpmsg_activate(FAR struct sensor_lowerhalf_s *lower,
                                 FAR struct file *filep,
                                 bool enable)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;

  if (drv->ops->activate)
    {
      return drv->ops->activate(drv, filep, enable);
    }

  return 0;
}

SENSOR_RPMSG_FUNCTION(set_interval, SNIOC_SET_INTERVAL,
                      *interval, interval, 0, false, uint32_t)
SENSOR_RPMSG_FUNCTION(batch, SNIOC_BATCH, *latency, latency, 0,
                      false, uint32_t)
SENSOR_RPMSG_FUNCTION(selftest, SNIOC_SELFTEST, arg, arg, 0,
                      true, unsigned long)
SENSOR_RPMSG_FUNCTION(set_calibvalue, SNIOC_SET_CALIBVALUE,
                      arg, arg, 256, true, unsigned long)
SENSOR_RPMSG_FUNCTION(calibrate, SNIOC_CALIBRATE, arg, arg, 256,
                      true, unsigned long)

static int sensor_rpmsg_flush(FAR struct sensor_lowerhalf_s *lower,
                              FAR struct file *filep)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  int ret = -ENOTTY;

  if (drv->ops->flush)
    {
      ret = drv->ops->flush(drv, filep);
    }
  else if ((filep->f_oflags & SENSOR_REMOTE) ||
           dev->nadvertisers > 0)
    {
      /* If the driver (drv) does not support the flush operation,
       * and the caller is a remote invocation or the current device
       * is an advertiser, then you can still consider the flush
       * operation as unsupported and therefore return ENOTSUP
       */

      return -ENOTSUP;
    }
  else
    {
      sminfo(dev->name, "rpmsg flushing");
      ret = sensor_rpmsg_ioctl(dev, SNIOC_FLUSH, 0, 0, true);
      sminfo(dev->name, "rpmsg flush send done, ret:%d", ret);
    }

  return ret;
}

static int sensor_rpmsg_set_wakeup(FAR struct sensor_lowerhalf_s *lower,
                                   FAR struct file *filep,
                                   bool wakeup)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  int ret = -ENOTTY;

  sminfo(dev->name, "rpsmg set wakeup status:%d", wakeup);
  if (drv->ops->set_wakeup)
    {
      ret = drv->ops->set_wakeup(drv, filep, wakeup);
    }
  else if ((filep->f_oflags & SENSOR_REMOTE) ||
           dev->nadvertisers > 0)
    {
      /* If the driver (drv) does not support the wakeup operation,
       * and the caller is a remote invocation or the current device
       * is an advertiser, then you can still consider the wakeup
       * operation as unsupported and therefore return ENOTSUP
       */

      return -ENOTSUP;
    }
  else
    {
      sminfo(dev->name, "rpsmg set wakeup to remote core:%d", wakeup);
      ret = sensor_rpmsg_ioctl(dev, SNIOC_SET_WAKEUP,
                               wakeup, 0, true);
    }

  return ret;
}

static int sensor_rpmsg_get_info(FAR struct sensor_lowerhalf_s *lower,
                                 FAR struct file *filep,
                                 FAR struct sensor_device_info_s *info)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  int ret = -ENOTTY;

  if (drv->ops->get_info)
    {
      ret = drv->ops->get_info(drv, filep, info);
    }

  if (ret == -ENOTTY && !(filep->f_oflags & SENSOR_REMOTE))
    {
      ret = sensor_rpmsg_ioctl(dev, SNIOC_GET_INFO,
                               (unsigned long)(uintptr_t)info,
                               sizeof(struct sensor_device_info_s), true);
    }

  return ret;
}

static int sensor_rpmsg_control(FAR struct sensor_lowerhalf_s *lower,
                                FAR struct file *filep,
                                int cmd, unsigned long arg)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;
  FAR struct sensor_lowerhalf_s *drv = dev->drv;
  FAR struct sensor_ioctl_s *ioctl = (FAR void *)(uintptr_t)arg;
  int ret = -ENOTTY;

  if (drv->ops->control)
    {
      ret = drv->ops->control(drv, filep, cmd, arg);
    }

  if (ret == -ENOTTY && !(filep->f_oflags & SENSOR_REMOTE) &&
      _SNIOCVALID(cmd))
    {
      return sensor_rpmsg_ioctl(dev, cmd, arg,
                                sizeof(*ioctl) + ioctl->len, true);
    }

  return ret;
}

static void sensor_rpmsg_data_worker(FAR void *arg)
{
  FAR struct sensor_rpmsg_ept_s *sre = arg;
  int ret;

  nxrmutex_lock(&sre->lock);
  if (sre->buffer)
    {
      ret = rpmsg_send_nocopy(&sre->ept, sre->buffer, sre->written);
      if (ret < 0)
        {
          rpmsg_release_tx_buffer(&sre->ept, sre->buffer);
          snerr("ERROR: push event rpmsg send failed:%d, %s\n",
                ret, rpmsg_get_cpuname(sre->ept.rdev));
        }

      sre->buffer = NULL;
    }

  nxrmutex_unlock(&sre->lock);
}

static void sensor_rpmsg_push_event_one(FAR struct sensor_rpmsg_dev_s *dev,
                                        FAR struct sensor_rpmsg_stub_s *stub,
                                        bool flushed)
{
  FAR struct sensor_rpmsg_cell_s *cell;
  FAR struct sensor_rpmsg_ept_s *sre;
  FAR struct sensor_rpmsg_data_s *msg;
  struct sensor_ustate_s state;
  uint64_t now;
  bool updated;
  int ret;

  /* If stub is non wakeup and remote cpu is sleep, don't send data
   * to remote
   */

  sre = container_of(stub->ept, struct sensor_rpmsg_ept_s, ept);
  if (!stub->wakeup && !rpmsg_is_running(sre->ept.rdev))
    {
      return;
    }

  /* Get state of device to do send data with timeout */

  ret = file_ioctl(&stub->file, SNIOC_GET_USTATE, &state);
  if (ret < 0)
    {
      return;
    }

  if (state.interval == UINT32_MAX)
    {
      state.interval = 0;
    }

  nxrmutex_lock(&sre->lock);

  /* Cancel work to fill new data to buffer */

  if (sre->buffer)
    {
      work_cancel(HPWORK, &sre->work);
    }

  for (; ; )
    {
      ret = file_ioctl(&stub->file, SNIOC_UPDATED, &updated);
      if (ret < 0 || (!updated && !flushed))
        {
          break;
        }

      /* If buffer isn't created or it doesn't have enough space to fill
       * new data, you should create or send this buffer at once.
       */

      if (!sre->buffer ||
          sre->written + sizeof(*cell) + state.esize > sre->space)
        {
          if (sre->buffer)
            {
              rpmsg_send_nocopy(&sre->ept, sre->buffer, sre->written);
              sre->buffer = NULL;
            }

          msg = rpmsg_get_tx_payload_buffer(&sre->ept, &sre->space, true);
          sre->buffer = msg;
          if (!msg)
            {
              snerr("ERROR: push event get buffer failed:%s\n",
                    rpmsg_get_cpuname(sre->ept.rdev));
              nxrmutex_unlock(&sre->lock);
              return;
            }

          msg->command = SENSOR_RPMSG_PUBLISH;
          sre->written = sizeof(*msg);
          sre->expire  = UINT64_MAX;
        }

      cell = sre->buffer + sre->written;
      if (flushed)
        {
          flushed = false;
          stub->flushing = false;
        }
      else
        {
          ret  = file_read(&stub->file, cell->data,
                           sre->space - sre->written - sizeof(*cell));
          if (ret <= 0)
            {
              break;
            }
        }

      smdebug(dev->name, "rpmsg push event, "
              "readcount:%d to remote:%s i:%" PRIu32 ", l:%" PRIu32,
              ret, rpmsg_get_cpuname(sre->ept.rdev),
              state.interval, state.latency);
      cell->len     = ret;
      cell->cookie  = stub->cookie;
      cell->nbuffer = dev->lower.nbuffer;

      sre->written += (sizeof(*cell) + ret + 0x7) & ~0x7;
    }

  /* If buffer timeout is expired, do rpmsg_send_nocopy, otherwise using
   * delay work to send data.
   */

  now = sensor_get_timestamp();
  if (sre->expire <= now && sre->buffer)
    {
      ret = rpmsg_send_nocopy(&sre->ept, sre->buffer, sre->written);
      if (ret < 0)
        {
          rpmsg_release_tx_buffer(&sre->ept, sre->buffer);
          snerr("ERROR: push event rpmsg send failed:%d, %s\n",
                ret, rpmsg_get_cpuname(sre->ept.rdev));
        }

      sre->buffer = NULL;
    }
  else
    {
      if (sre->expire == UINT64_MAX ||
          sre->expire - now > state.interval / 2)
        {
          sre->expire = now + state.interval / 2;
        }

      work_queue(HPWORK, &sre->work, sensor_rpmsg_data_worker, sre,
                 (sre->expire - now) / USEC_PER_TICK);
    }

  nxrmutex_unlock(&sre->lock);
}

static ssize_t sensor_rpmsg_push_event(FAR void *priv, FAR const void *data,
                                       size_t bytes)
{
  FAR struct sensor_rpmsg_dev_s *dev = priv;
  FAR struct sensor_rpmsg_stub_s *stub;
  FAR struct sensor_rpmsg_stub_s *stmp;
  ssize_t ret;

  /* Push new data to upperhalf driver's circular buffer */

  ret = dev->push_event(dev->upper, data, bytes);
  if (ret < 0)
    {
      return ret;
    }

  /* Send new data to own proxy(remote subscribers), don't care whether
   * is successful, and must return length of written.
   */

  sensor_rpmsg_lock(dev);
  list_for_every_entry_safe(&dev->stublist, stub, stmp,
                            struct sensor_rpmsg_stub_s, node)
    {
      sensor_rpmsg_push_event_one(dev, stub,
                                  stub->flushing && bytes == 0);
    }

  sensor_rpmsg_unlock(dev);
  return ret;
}

static void
sensor_rpmsg_notify_interval(FAR void *priv, uint32_t interval)
{
  FAR struct sensor_rpmsg_dev_s *dev = priv;

  dev->notify_interval(dev->upper, interval);
}

static FAR struct sensor_rpmsg_dev_s *
sensor_rpmsg_find_dev(FAR const char *path)
{
  FAR struct sensor_rpmsg_dev_s *dev;

  down_read(&g_dev_lock);
  list_for_every_entry(&g_devlist, dev, struct sensor_rpmsg_dev_s, node)
    {
      if (strcmp(dev->path, path) == 0)
        {
          up_read(&g_dev_lock);
          return dev;
        }
    }

  up_read(&g_dev_lock);
  return NULL;
}

static void sensor_rpmsg_update_config(FAR struct sensor_rpmsg_dev_s *dev,
                                       FAR struct sensor_state_s *state)
{
  if (state->esize == UINT32_MAX)
    {
      /* Invalid esize of state */

      return;
    }

  /* sync interval and latency */

  if (state->min_interval != UINT32_MAX)
    {
      sensor_rpmsg_ioctl(dev, SNIOC_SET_INTERVAL, state->min_interval,
                         0, false);
    }

  if (state->min_latency != UINT32_MAX)
    {
      sensor_rpmsg_ioctl(dev, SNIOC_BATCH, state->min_latency, 0, false);
    }

  if (state->wakeup)
    {
      sensor_rpmsg_ioctl(dev, SNIOC_SET_WAKEUP, state->wakeup, 0, false);
    }
}

static int sensor_rpmsg_adv_handler(FAR struct rpmsg_endpoint *ept,
                                    FAR void *data, size_t len,
                                    uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;
  FAR struct sensor_rpmsg_proxy_s *proxy;
  FAR struct sensor_rpmsg_dev_s *dev;
  struct sensor_state_s state;
  int ret;

  dev = sensor_rpmsg_find_dev(msg->path);
  if (!dev || !dev->nsubscribers)
    {
      return 0;
    }

  state.esize = UINT32_MAX;
  proxy = sensor_rpmsg_alloc_proxy(dev, ept, msg, &state);
  if (!proxy)
    {
      snerr("ERROR: adv create proxy failed:%s\n", dev->path);
    }
  else
    {
      msg->cookie = (uint64_t)(uintptr_t)dev;
      msg->command = SENSOR_RPMSG_ADVERTISE_ACK;
      ret = rpmsg_send(ept, msg, len);
      if (ret < 0)
        {
          snerr("ERROR: adv rpmsg send failed:%s, %d, %s\n",
                dev->path, ret, rpmsg_get_cpuname(ept->rdev));
        }
      else
        {
          sensor_rpmsg_update_config(dev, &state);
          sminfo(dev->name, "rpmsg adv proxy success, remote:%s",
                 rpmsg_get_cpuname(ept->rdev));
        }
    }

  return 0;
}

static int sensor_rpmsg_advack_handler(FAR struct rpmsg_endpoint *ept,
                                       FAR void *data, size_t len,
                                       uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;
  FAR struct sensor_rpmsg_dev_s *dev;

  dev = sensor_rpmsg_find_dev(msg->path);
  if (dev && !sensor_rpmsg_alloc_stub(dev, ept, msg->cookie))
    {
      snerr("ERROR: advack failed:%s, %s\n", dev->path,
            rpmsg_get_cpuname(ept->rdev));
    }

  return 0;
}

static int sensor_rpmsg_unadv_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv)
{
  return 0;
}

static int sensor_rpmsg_sub_handler(FAR struct rpmsg_endpoint *ept,
                                    FAR void *data, size_t len,
                                    uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;
  FAR struct sensor_rpmsg_dev_s *dev;
  FAR struct sensor_rpmsg_stub_s *stub;
  int ret;

  dev = sensor_rpmsg_find_dev(msg->path);

  /* The received subscription broadcast is not processed under the
   * following conditions:
   *   1.There is no node under the current core.(/dev/uorb/xxx)
   *   2.The current core has not advertised any data. nadvertisers is -1.
   *   3.When the number of advertisers is 0, it is not persistent attribute.
   */

  if (!dev || dev->nadvertisers < 0 || (dev->nadvertisers == 0 &&
      !dev->lower.persist))
    {
      return 0;
    }

  stub = sensor_rpmsg_alloc_stub(dev, ept, msg->cookie);
  if (!stub)
    {
      snerr("ERROR: sub alloc stub failed:%s\n", dev->path);
    }
  else
    {
      msg->cookie  = (uint64_t)(uintptr_t)dev;
      msg->command = SENSOR_RPMSG_SUBSCRIBE_ACK;
      msg->nbuffer = dev->lower.nbuffer;
      msg->persist = dev->lower.persist;
      ret = rpmsg_send(ept, msg, len);
      if (ret < 0)
        {
          sensor_rpmsg_lock(dev);
          sensor_rpmsg_free_stub(stub);
          sensor_rpmsg_unlock(dev);
          snerr("ERROR: sub rpmsg send failed:%s, %d, %s\n",
                dev->path, ret, rpmsg_get_cpuname(ept->rdev));
        }
    }

  return 0;
}

static int sensor_rpmsg_suback_handler(FAR struct rpmsg_endpoint *ept,
                                       FAR void *data, size_t len,
                                       uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;
  FAR struct sensor_rpmsg_dev_s *dev;
  struct sensor_state_s state;

  state.esize = UINT32_MAX;
  dev = sensor_rpmsg_find_dev(msg->path);
  if (dev && (!dev->nsubscribers ||
      !sensor_rpmsg_alloc_proxy(dev, ept, msg, &state)))
    {
      sensor_rpmsg_advsub_one(dev, ept, SENSOR_RPMSG_UNSUBSCRIBE);
      snerr("ERROR: suback failed:%s\n", dev->path);
    }
  else
    {
      sensor_rpmsg_update_config(dev, &state);
      sminfo(dev->name, "rpmsg suback success, remote:%s",
             rpmsg_get_cpuname(ept->rdev));
    }

  return 0;
}

static int sensor_rpmsg_unsub_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;
  FAR struct sensor_rpmsg_dev_s *dev;
  FAR struct sensor_rpmsg_stub_s *stub;

  dev = sensor_rpmsg_find_dev(msg->path);
  if (!dev)
    {
      return 0;
    }

  sensor_rpmsg_lock(dev);
  list_for_every_entry(&dev->stublist, stub,
                       struct sensor_rpmsg_stub_s, node)
    {
      if (stub->ept == ept && stub->cookie == msg->cookie)
        {
          sensor_rpmsg_free_stub(stub);
          break;
        }
    }

  sensor_rpmsg_unlock(dev);
  return 0;
}

static int sensor_rpmsg_publish_handler(FAR struct rpmsg_endpoint *ept,
                                        FAR void *data, size_t len,
                                        uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_data_s *msg = data;
  FAR struct sensor_rpmsg_cell_s *cell;
  FAR struct sensor_rpmsg_stub_s *stub;
  FAR struct sensor_rpmsg_stub_s *stmp;
  FAR struct sensor_rpmsg_dev_s *dev;
  size_t written = sizeof(*msg);

  while (written < len)
    {
      cell = (FAR struct sensor_rpmsg_cell_s *)
             ((FAR char *)data + written);
      dev = (FAR struct sensor_rpmsg_dev_s *)(uintptr_t)cell->cookie;
      if (cell->nbuffer > dev->lower.nbuffer)
        {
          struct file file;
          int ret;

          ret = file_open(&file, dev->path, SENSOR_REMOTE | O_CLOEXEC);
          if (ret >= 0)
            {
              ret = file_ioctl(&file, SNIOC_SET_BUFFER_NUMBER,
                               cell->nbuffer);
              if (ret >= 0)
                {
                  dev->lower.nbuffer = cell->nbuffer;
                }

              file_close(&file);
            }
        }

      dev->push_event(dev->upper, cell->data, cell->len);

      /* When the remote core publishes a message, the subscribed cores will
       * receive the message. When the subscribed core publishes a new
       * message, it will take away the message published by the remote core,
       * so all stublist information needs to be updated.
       */

      sensor_rpmsg_lock(dev);
      list_for_every_entry_safe(&dev->stublist, stub, stmp,
                                struct sensor_rpmsg_stub_s, node)
        {
          file_read(&stub->file, NULL, cell->len);
        }

      sensor_rpmsg_unlock(dev);

      smdebug(dev->name, "rpmsg receive data: cnt:%" PRIu32 ", "
              "from remote:%s", cell->len, rpmsg_get_cpuname(ept->rdev));
      written += sizeof(*cell) + cell->len + 0x7;
      written &= ~0x7;
    }

  return 0;
}

static int sensor_rpmsg_ioctl_handler(FAR struct rpmsg_endpoint *ept,
                                      FAR void *data, size_t len,
                                      uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_ioctl_s *msg = data;
  FAR struct sensor_rpmsg_stub_s *stub;
  FAR struct sensor_rpmsg_stub_s *stmp;
  FAR struct sensor_rpmsg_dev_s *dev;
  unsigned long arg;
  int ret;

  arg = msg->arglen > 0 ? (unsigned long)(uintptr_t)msg->arg :
                          *(FAR uint64_t *)msg->arg;
  dev = (FAR struct sensor_rpmsg_dev_s *)(uintptr_t)msg->proxy;
  sensor_rpmsg_lock(dev);
  list_for_every_entry_safe(&dev->stublist, stub, stmp,
                            struct sensor_rpmsg_stub_s, node)
    {
      if (stub->ept == ept)
        {
          msg->result = file_ioctl(&stub->file, msg->request, arg);
          if (msg->result >= 0 && msg->request == SNIOC_FLUSH)
            {
              sminfo(dev->name, "receiving flush request");
              stub->flushing = true;
            }

          if (msg->request == SNIOC_SET_WAKEUP)
            {
              stub->wakeup = arg;
              sminfo(dev->name, "receiving wakeup update:%d", stub->wakeup);
            }

          if (msg->cookie)
            {
              msg->command = SENSOR_RPMSG_IOCTL_ACK;
              ret = rpmsg_send(ept, msg, len);
              if (ret < 0)
                {
                  snerr("ERROR: ioctl rpmsg send failed:%s, %d, %s\n",
                        dev->path, ret, rpmsg_get_cpuname(ept->rdev));
                }
            }

          break;
        }
    }

  sensor_rpmsg_unlock(dev);
  return 0;
}

static int sensor_rpmsg_ioctlack_handler(FAR struct rpmsg_endpoint *ept,
                                         FAR void *data, size_t len,
                                         uint32_t src, FAR void *priv)
{
  FAR struct sensor_rpmsg_ioctl_cookie_s *cookie;
  FAR struct sensor_rpmsg_ioctl_s *msg = data;

  cookie = (FAR struct sensor_rpmsg_ioctl_cookie_s *)
           (uintptr_t)msg->cookie;
  cookie->result = msg->result;
  if (msg->result >= 0 && msg->arglen > 0)
    {
      memcpy(cookie->data, msg->arg, msg->arglen);
    }

  rpmsg_post(ept, &cookie->sem);
  return 0;
}

static int sensor_rpmsg_ept_cb(FAR struct rpmsg_endpoint *ept,
                               FAR void *data, size_t len, uint32_t src,
                               FAR void *priv)
{
  FAR struct sensor_rpmsg_advsub_s *msg = data;

  if (msg->command < sizeof(g_sensor_rpmsg_handler) /
                     sizeof(g_sensor_rpmsg_handler[0]))
    {
      return g_sensor_rpmsg_handler[msg->command](ept, data, len, src, priv);
    }

  return -EINVAL;
}

static void sensor_rpmsg_bound_worker(FAR void *arg)
{
  FAR struct sensor_rpmsg_ept_s *sre = arg;
  FAR struct sensor_rpmsg_dev_s *dev;

  down_read(&g_dev_lock);
  list_for_every_entry(&g_devlist, dev,
                       struct sensor_rpmsg_dev_s, node)
    {
      sensor_rpmsg_lock(dev);
      if (dev->nadvertisers > 0)
        {
          sensor_rpmsg_advsub_one(dev, &sre->ept, SENSOR_RPMSG_ADVERTISE);
        }

      if (dev->nsubscribers > 0)
        {
          sensor_rpmsg_advsub_one(dev, &sre->ept, SENSOR_RPMSG_SUBSCRIBE);
        }

      sensor_rpmsg_unlock(dev);
    }

  up_read(&g_dev_lock);
}

static void sensor_rpmsg_device_ns_bound(FAR struct rpmsg_endpoint *ept)
{
  FAR struct sensor_rpmsg_ept_s *sre;

  sre = container_of(ept, struct sensor_rpmsg_ept_s, ept);

  down_write(&g_ept_lock);
  list_add_tail(&g_eptlist, &sre->node);
  up_write(&g_ept_lock);

  /* Broadcast all device to ready ept in work context */

  work_queue(HPWORK, &sre->bound_work, sensor_rpmsg_bound_worker, sre, 0);
}

static void sensor_rpmsg_ept_release(FAR struct rpmsg_endpoint *ept)
{
  FAR struct sensor_rpmsg_ept_s *sre;
  FAR struct sensor_rpmsg_dev_s *dev;
  FAR struct sensor_rpmsg_stub_s *stub;
  FAR struct sensor_rpmsg_proxy_s *proxy;

  sre = container_of(ept, struct sensor_rpmsg_ept_s, ept);

  /* Remove all proxy and stub info in sensor device with the ept
   * destoryed.
   */

  down_read(&g_dev_lock);
  list_for_every_entry(&g_devlist, dev,
                       struct sensor_rpmsg_dev_s, node)
    {
      sensor_rpmsg_lock(dev);
      list_for_every_entry(&dev->proxylist, proxy,
                           struct sensor_rpmsg_proxy_s, node)
        {
          if (proxy->ept == ept)
            {
              sensor_rpmsg_free_proxy(dev, proxy);
              break;
            }
        }

      list_for_every_entry(&dev->stublist, stub,
                           struct sensor_rpmsg_stub_s, node)
        {
          if (stub->ept == ept)
            {
              sensor_rpmsg_free_stub(stub);
              break;
            }
        }

      sensor_rpmsg_unlock(dev);
    }

  up_read(&g_dev_lock);

  down_write(&g_ept_lock);
  if (list_in_list(&sre->node))
    {
      list_delete(&sre->node);
    }

  up_write(&g_ept_lock);

  nxrmutex_destroy(&sre->lock);
  kmm_free(sre);
}

static void sensor_rpmsg_device_created(FAR struct rpmsg_device *rdev,
                                        FAR void *priv)
{
  FAR struct sensor_rpmsg_ept_s *sre;

  sre = kmm_zalloc(sizeof(*sre));
  if (!sre)
    {
      return;
    }

  sre->ept.priv = sre;
  nxrmutex_init(&sre->lock);
  sre->ept.ns_bound_cb = sensor_rpmsg_device_ns_bound;
  sre->ept.release_cb = sensor_rpmsg_ept_release;

  if (rpmsg_create_ept(&sre->ept, rdev, SENSOR_RPMSG_EPT_NAME,
                       RPMSG_ADDR_ANY, RPMSG_ADDR_ANY,
                       sensor_rpmsg_ept_cb,
                       rpmsg_destroy_ept) < 0)
    {
      nxrmutex_destroy(&sre->lock);
      kmm_free(sre);
    }
}

/****************************************************************************
 * Public Functions
 ****************************************************************************/

/****************************************************************************
 * Name: sensor_rpmsg_register
 *
 * Description:
 *   This function registers rpmsg takeover for the real lower half, and
 *   initialize rpmsg resource.
 *
 * Input Parameters:
 *   lower - The instance of lower half sensor driver.
 *   path  - The path of character node, ex: /dev/uorb/xxx.
 *
 * Returned Value:
 *   The takeover rpmsg lowerhalf returned on success, NULL on failure.
 *
 ****************************************************************************/

FAR struct sensor_lowerhalf_s *
sensor_rpmsg_register(FAR struct sensor_lowerhalf_s *lower,
                      FAR const char *path)
{
  FAR struct sensor_rpmsg_ept_s *sre;
  FAR struct sensor_rpmsg_dev_s *dev;
  size_t size;

  if (lower->ops->fetch)
    {
      return lower;
    }

  size = strlen(path);
  dev = kmm_zalloc(sizeof(*dev) + size);
  if (!dev)
    {
      return NULL;
    }

  /* Initialize the sensor rpmsg device structure */

  nxsem_init(&dev->proxysem, 0, 0);
  list_initialize(&dev->stublist);
  list_initialize(&dev->proxylist);
  strlcpy(dev->path, path, size + 1);

  dev->name              = basename(dev->path);
  dev->nadvertisers      = lower->ops->activate ? 1 : -1;
  dev->push_event        = lower->push_event;
  dev->notify_interval   = lower->notify_interval;
  dev->upper             = lower->priv;
  lower->push_event      = sensor_rpmsg_push_event;
  lower->notify_interval = sensor_rpmsg_notify_interval;
  lower->priv            = dev;
  memcpy(&dev->lower, lower, sizeof(*lower));
  dev->lower.ops         = &g_sensor_rpmsg_ops;
  dev->drv               = lower;

  /* If openamp is ready, send advertisement to remote proc */

  down_write(&g_dev_lock);
  list_add_tail(&g_devlist, &dev->node);
  up_write(&g_dev_lock);
  if (lower->ops->activate)
    {
      down_read(&g_ept_lock);
      list_for_every_entry(&g_eptlist, sre, struct sensor_rpmsg_ept_s,
                           node)
        {
          sensor_rpmsg_advsub_one(dev, &sre->ept, SENSOR_RPMSG_ADVERTISE);
        }

      up_read(&g_ept_lock);
    }

  return &dev->lower;
}

/****************************************************************************
 * Name: sensor_rpmsg_unregister
 *
 * Description:
 *   This function unregisters rpmsg takeover for the real lower half, and
 *   release rpmsg resource. This API corresponds to sensor_rpmsg_register.
 *
 * Input Parameters:
 *   lower - The instance of lower half sensor driver.
 *
 * Returned Value:
 *   The takeover rpmsg lowerhalf returned on success, NULL on failure.
 ****************************************************************************/

FAR struct sensor_lowerhalf_s *
sensor_rpmsg_unregister(FAR struct sensor_lowerhalf_s *lower)
{
  FAR struct sensor_rpmsg_dev_s *dev = lower->priv;

  if (dev->nsubscribers != 0)
    {
      return NULL;
    }

  down_write(&g_dev_lock);
  list_delete(&dev->node);
  up_write(&g_dev_lock);

  nxsem_destroy(&dev->proxysem);
  lower->priv = dev->upper;
  kmm_free(dev);
  return lower;
}

/****************************************************************************
 * Name: sensor_rpmsg_initialize
 *
 * Description:
 *   This function initializes the context of sensor rpmsg, registers
 *   rpmsg callback and prepares enviroment to intercat with remote sensor.
 *
 * Returned Value:
 *   OK on success; A negated errno value is returned on any failure.
 *
 ****************************************************************************/

int sensor_rpmsg_initialize(void)
{
  return rpmsg_register_callback(NULL, sensor_rpmsg_device_created,
                                 NULL, NULL, NULL);
}