VGX Server Multi-Node Setup

import pyvgx


def prepare( request:pyvgx.PluginRequest, query:str ):
  """
  Dispatcher pre-processor: look for cached query response
  and return immediately if cached, otherwise forward
  request to back-end
  """
  response = get_cached( query )
  if response:
    return response
  return request


def complete( response:pyvgx.PluginResponse ):
  """
  Dispatcher post-processor: format the final merged
  response as a list of strings
  """
  items = []
  for score, item in response.entries:
    items.append( "{:.4f}: {}".format( score, transform(item) ) )
  return items


def search( request:pyvgx.PluginRequest, query:str ):
  """
  Execute search and return response object
  """
  response = pyvgx.PluginResponse( sortby=pyvgx.S_RANK )
  for score, item in run_query( query ):
    response.Append( score, item )
  return response


PluginDefs = [
  # Use locally defined plugin functions
  {
    "name":    "PluginFromLocalFunctions",
    "engine":  search,
    "pre":     prepare,
    "post":    complete
  },
  # Use plugin functions defined in class Example
  # within module example.plugin_module
  {
    "name":    "PluginFromModuleFunctions",
    "package": "example",
    "module":  ".plugin_module",
    "engine":  "Example.Engine",
    "pre":     "Example.PreProcessor",
    "post":    "Example.PostProcessor",
  }
]

# Now use PluginDefs with
# pyvgx.VGXInstance.StartInstance( plugins=PluginDefs )

import pyvgx

def sqrt( request, x:float ):
  return x ** 0.5

PLUGINS = [
    {
        "name": "SquareRoot",
        "engine": sqrt
    }
]

DESCRIPTOR = {
  "name": "Square Root Service",
  "instances": {
    "SQRT01": {
      "type": "generic",
      "host":"127.0.0.1",
      "hport":9500
    }
  },
  "topology": {
    "transaction":{
      "SQRT01": {}
    },
    "dispatch": {
    }
  },
  "graphs":[]
}

pyvgx.initadmin()

instance = pyvgx.VGXInstance.StartInstance(
  id="SQRT01",
  descriptor=DESCRIPTOR,
  plugins=PLUGINS
)

pyvgx.system.RunServer( name=instance.id )

{
    "name": "Example Service",

    "instances": {
        "A1": {
            "type": "admin", "description": "Admin",
            "group": 1000.001,
            "hport": 9000
        },
        "TD": {
            "type": "dispatch", "description": "Top Dispatcher",
            "group": 2000.001,
            "hport": 9990
        },
        "B0.1": {
            "type": "builder", "description": "Builder 0.1",
            "hport": 9100, "tport": 10100
        },
        "S1.1": {
            "type": "search", "description": "Search 1.1",
            "hport": 9110, "tport": 10110
        },
        "S2.1": {
            "type": "search", "description": "Search 2.1",
            "hport": 9120, "tport": 10120
        },
        "B0.2": {
            "type": "builder", "description": "Builder 0.2",
            "hport": 9200, "tport": 10200
        },
        "S1.2": {
            "type": "search", "description": "Search 1.2",
            "hport": 9210, "tport": 10210
        },
        "S2.2": {
            "type": "search", "description": "Search 2.2",
            "hport": 9220, "tport": 10220
        }
    },

    "common": {
        "host": "127.0.0.1"
    },

    "topology": {
        "transaction": {
            "A1": {},
            "B0.1": [
                "S1.1",
                "S2.1"
            ],
            "B0.2": [
                "S1.2",
                "S2.2"
            ]
        },
        "dispatch": {
            "TD": [
                { "channels": 32, "priority": 1,
                  "partitions": [ "S1.1", "S1.2" ]
                },
                { "channels": 32, "priority": 1,
                  "partitions": [ "S2.1", "S2.2" ]
                },
                { "channels": 2, "priority": 20,
                  "partitions": [ "B0.1", "B0.2" ],
                  "primary": 1
                }
            ]
        }
    },

    "graphs": ["g1"]
}

from pyvgx import *
import pyvgx
import sys
import time


def LocalSamplePlugin( request:PluginRequest ):
    """
    Hello world plugin
    """
    response = PluginResponse()
    msg = "Hello, the current time is: {}".format( time.ctime() )
    response.Append( msg )
    return response



def GetPluginDefinitions():
    """
    Service endpoint definitions
    """
    return [
      {
        "name"   : "search",         # /vgx/plugin/search
        "package": __package__,      # 'service'
        "module" : ".exampleplugin", # exampleplugin.py
        "engine" : "Search",         # name of plugin function
        "pre"    : None,             # no pre-processor
        "post"   : None,             # no post-processor
        "graph"  : "*"               # use 'g1' defined in vgx.cf
      },
      {
        "name"   : "add",            # /vgx/plugin/add
        "package": __package__,      # 'service'
        "module" : ".exampleplugin", # exampleplugin.py
        "engine" : "Add",            # name of plugin function
        "pre"    : "PreAdd",         # name of pre-processor
        "post"   : None,             # no post-processor
        "graph"  : "*"               # use 'g1' defined in vgx.cf
      },
      {
        "name"   : "hello",          # /vgx/plugin/hello
        "engine" : LocalSamplePlugin
      }
    ]


def AdminInit():
    """
    Finalize startup of all instances
    """
    t0 = time.time()
    expected_ids = set(
      pyvgx.Descriptor( "vgx.cf" ).instances.keys()
    )
    while time.time() - t0 < 30:
        try:
            S = pyvgx.VGXAdmin.Run(
                arguments=[ '--status', '*' ],
                address="127.0.0.1:9001",
                print_to_stdout=False,
                default_descriptor_filename="vgx.cf"
            )
            ids = set( [x.split()[0] for x in S[0]['trace']
                        if x.split()[1] == '0d']
            )
            assert ids == expected_ids
            R = pyvgx.VGXAdmin.Run(
                arguments=[ '--attach', 'B*',
                            '--servicein', 'S*'
                ],
                address="127.0.0.1:9001",
                print_to_stdout=False,
                default_descriptor_filename="vgx.cf"
            )
            for r in R[1:]:
                A = ['attached', 'service_in']
                assert r[1]['action'] in A
        except:
            time.sleep(1)



def RunService( instance_id ):
    """
    Run VGX Server instance
    """
    # Initialize core library to enable instance startup
    pyvgx.initadmin()

    # Service endpoint definitions
    plugins = GetPluginDefinitions()

    # VGX System descriptor
    descriptor = pyvgx.VGXInstance.GetDescriptor( "vgx.cf" )

    # Start VGX instance
    instance = pyvgx.VGXInstance.StartInstance(
        id         = instance_id,
        descriptor = descriptor,
        basedir    = "demoservice",
        plugins    = plugins
    )

    # Finalize startup
    if instance.type == "admin":
        AdminInit()

    # Run until SIGINT
    pyvgx.system.RunServer( name=instance.description )



if __name__ == "__main__":
    RunService( sys.argv[1] )

from pyvgx import *
import random
import time


def Search( request:PluginRequest,
            graph:Graph,
            name:str,
            hits:int=10,
            sortby:str="val",
            sortdir:str="desc",
            fields:int=F_AARC|F_RANK ) -> PluginResponse:
    """
    Engine plugin example: Search
    """
    root = system.Root()
    if sortdir == "desc":
        d = S_DESC
    else:
        d = S_ASC

    if sortby == "val":
        pr = PluginResponse( maxhits=hits, sortby=S_VAL|d )
        sf = F_VAL
        fn = "arc"
        fn2 = "value"
    elif sortby == "id":
        pr = PluginResponse( maxhits=hits, sortby=S_ID|d )
        sf = F_ID
        fn = "id"
    elif sortby == "deg":
        pr = PluginResponse( maxhits=hits, sortby=S_DEG|d )
        sf = F_DEG
        fn = "degree"
    else:
        pr = PluginResponse( maxhits=hits, sortby=S_NONE )
        sf = 0
        fn = None

    try:
        result = graph.Neighborhood(
            id      = name,
            arc     = D_ANY,
            hits    = pr.maxhits,
            sortby  = pr.sortby,
            fields  = fields|sf,
            result  = R_DICT,
            timeout = 1000
        )
        pre_message = request.get('pre-message')
        for r in result:
            rv = r.get(fn)
            if type(rv) is dict:
                rv = rv.get(fn2)
            r['pre'] = pre_message
            r['this'] = root
            pr.Append( rv, r )
    except KeyError:
        return pr
    except Exception as err:
        pr.message = repr(err)
        print(err)
    return pr



def PreAdd( request:PluginRequest, graph:Graph ) -> PluginRequest:
    """
    Pre-processor plugin example: PreAdd
    """
    # Always route this request to the primary row
    request.primary = 1
    return request



def Add( request:PluginRequest,
         graph:Graph,
         N:int=10000,
         count:int=500,
         sleep:int=0 ) -> PluginResponse:
    """
    Engine plugin example: Add
    """
    # Add random data
    response = PluginResponse()
    if count >= 0:
        for i in range(count):
            a = random.randint(1, N)
            while True:
                b = random.randint(1, N)
                if a != b:
                    break
            r = random.random()
            A = None
            B = None
            try:
                graph.CreateVertex( str(a) )
                graph.CreateVertex( str(b) )
                A, B = graph.OpenVertices(
                  [str(a), str(b)],
                  timeout=100
                )
                graph.Connect( A, ("to", M_FLT, r), B )
            finally:
                if A is not None:
                    A.Close()
                if B is not None:
                    B.Close()
            if sleep > 0:
                time.sleep( sleep/1000 )
        response.Append( "Added {}".format(count) )
    else:
        count = -count
        c0 = graph.Order()
        for i in range(count):
            if graph.Order() == 0:
                break
            x = graph.GetVertexID()
            graph.Disconnect( x, timeout=2000 )
            graph.DeleteVertex( x, timeout=2000 )
            if sleep > 0:
                time.sleep( sleep/1000 )
        c1 = graph.Order()
        response.Append( "Deleted {}".format( c0-c1 ) )
    return response

for %%i in (A1 TD B0.1 B0.2 S1.1 S1.2 S2.1 S2.2) do (
    start "" /MIN python -m service.exampleservice %%i
)

for i in A1 TD B0.1 B0.2 S1.1 S1.2 S2.1 S2.2; do
    python -m service.exampleservice "$i" > /dev/null 2>&1 &
done

vgxadmin --status "*"

curl "http://127.0.0.1:9990/vgx/plugin/add?N=250000&count=3000000"

curl "http://127.0.0.1:9990/vgx/plugin/search?name=12345"

vgxadmin --stop "*" --confirm