text:
THE VAULT arrives on DVD June 1st
Release dates: Now In Select Theaters, On Digital & On Demand
On DVD June 1, 2021
Jaume Balagueró
Rafa Martínez, Andrés Koppel, Borja Glez. Santaolalla, Michel Gaztambide, Rowan Athale
Freddie Highmore, Astrid Bergès-Frisbey, Sam Riley, Liam Cunningham, Jose Coronado, Luis Tosar, Emilio Gutiérrez Caba, Axel Stein, Daniel Holguín, with Famke Janssen
R for language
Action, Thriller, Crime
Freddie Highmore ("The Good Doctor") and Famke Janssen (X-Men) star in this globe-trotting, action-packed thriller following a high-risk heist. When an engineer (Highmore) learns of a mysterious, impenetrable fortress hidden under The Bank of Spain, he joins a crew of master thieves who plan to steal the legendary lost treasure locked inside while the whole country is distracted by Spain's World Cup Final. With thousands of soccer fans cheering in the streets, and security forces closing in, the crew have just minutes to pull off the score of a lifetime. Also starring Astrid Berges-Frisbey (Pirates of the Caribbean: On Stranger Tides) and Liam Cunningham ("Game of Thrones").
Posted by GenreOnline.net at 11:38 AM
The decades go by fast.
WILDCAT arrives on DVD June 29th
The Action-Packed, All-Star G.I. JOE Feature Films...
Newly Remastered Eddie Murphy Classics 48 Hours an...
UNDERCOVER PUNCH & GUN new trailer & on DIGTAL & B...
NIGHT OF THE SICARIO arrives on DVD June 22nd
Defending Jacob arrives on DVD July 6th
Leslie Lehr's Acclaimed Book, A Boob's Life Is Now...
SILAT WARRIOR: DEED OF DEATH, Zoom event with Dire...
MY FAIR LADY- 4K UHD TRAILER AVAILABLE NOW
Official 'Godzilla vs. Kong' Prequel Books  KINGDO...
Assault on VA-33- Out Now In Select Theaters, On D...
New Trailer for DELIVER US FROM EVIL, on Blu-ray &...
Cosmic Sin arrives on Blu-ray/DVD May 18th
Classic John Sturges Western Starring Kirk Douglas...

redpajama_set_name: RedPajamaCommonCrawl


text:
Official: Controlling wild horses in US will take $5b, 15 years
Wild horses are blamed for damage to federal lands in west, but conservationists say cattle and sheep are the culprits.
Wild horses roam free in parts of the US west. [File: Jim Urquhart/Reuters]
It will take $5bn and 15 years to get an over-population of wild horses under control on federal lands in the west of the United States, the acting head of the US Bureau of Land Management said on Wednesday, adding that several developments had made him more optimistic about his agency's ability to get the job done.
William Perry Pendley said the agency had adopted out more than 7,000 mustangs and burros captured last year, the most in 15 years and a 54 percent increase from the previous year.
He said more adoptions would help clear space in government-holding pens, so they can accelerate round-ups while scientists develop new fertility-control drugs to eventually shrink the size of the herds from 88,000 to the 27,000 he says the range can sustain.
He said a new coalition of animal welfare advocates and ranchers is helping promote solutions and Congress appears willing to help.
Pendley said the agency is in the process of hiring additional staff to accelerate round-ups in Nevada, the state with the most horses.
"I'm not going to speculate on what Congress is going to do about money," Pendley said. "But I know there is a sense of sincerity on the Hill about this issue. They get it."
Captured wild horses are kept in holding pens and put up for adoption. Some want the animals to be sold for slaughter. [File: Scott Sonner/AP Photo]
'Betrayal'
The Senate Appropriations Committee approved $35m last month for a new package of mustang proposals supported by an unprecedented alliance including the Humane Society of the United States, American Society for the Prevention of Cruelty to Animals, National Cattlemen's Beef Association and American Farm Bureau Federation.
They say it would eliminate the threat of slaughter for thousands of free-roaming horses and shrink the size of herds primarily through expanded fertility controls on the range and increased roundups in certain areas.
The proposal has been condemned by the largest and oldest mustang protection groups in the west, including the American Wild Horse Campaign and Friends of Animals.
"This proposal, which is really a betrayal by so-called wild horse advocates who are in bed with the meat industry, is management for extinction and putting money toward it is a step toward eradicating these iconic animals from our public lands," Friends of Animals President Priscilla Feral said in a statement on Wednesday.
In July, then-Acting BLM Director Casey Hammond said the Trump administration would not pursue lethal measures such as euthanasia or selling horses for slaughter.
But critics say the new plan could allow for sterilisation of mares.
They argue the animals must be permitted to roam the range in federally-protected management areas established under the Free-Roaming Horse and Burro Act of 1971.
They say BLM's population quotas are often outdated and lack scientific data to support roundups to cull herd sizes.
Helicopters are often used to round up wild horses that roam free across parts of the western US. [File: Jim Urquhart/Reuters]
Cattle vs wild horses
The $35m the Senate committee approved is part of the $35.8bn Interior Department appropriation bill. It's not clear when the full Senate will consider the measure.
Pendley said the agency is spending $50m a year to house 50,000 captured horses in government corrals and a further $30m in associated costs.
Alan Shepherd, the head of the bureau's wild horse and burro programme in Nevada, planned to join members of the new coalition at the Nevada Museum of Art in Reno on Wednesday night for the screening of a documentary that blames horses for the severe degradation of federal rangeland.
Doug Busselman, executive director of the Nevada Farm Bureau, said his group still wants Congress to allow the government to sell excess horses without the current ban on their resale for slaughter.
"As long as the numbers are so far above appropriate management levels, the tools of fertility control and adoption don't accomplish the needs for having wild horse and burro populations match the carrying capacity of the herd-management levels and the destruction of the habitat will continue," he said in a statement.
Conservationists say more damage is caused by the cattle and sheep that graze on public lands at a ratio of 15-to-1 mustangs.
"This film is a propaganda piece ... that scapegoats relatively rare wild horses for problems that in truth are caused by the domestic cattle that are widespread environmental problems across the West," said Erik Molvar, a wildlife biologist for the Western Watersheds Project.
SOURCE: AP news agency


FROM A DISTANCE, it looked as though a huge swarm of insects had just lifted off from the marsh. But as the boat drew closer, the myriad black specks grew larger and resolved into the distinctive shapes of birds: thousands of northern pintails, green-winged teal, northern shovelers and other ducks. "You're probably looking at 8,000 ducks," Ryan Lambert of Cajun Fishing Adventures said as he maneuvered his 28-foot, camouflaged craft through the labyrinth of a lush wetland off the Mississippi River near Buras, Louisiana.
The birds' late arrival "had a lot of duck hunters scratching their heads," says Ben Weber, sportsmen's outreach coordinator for Vanishing Paradise, a partnership of NWF, DU and Louisiana Wildlife Federation (an NWF affiliate) that works to protect and restore waterfowl habitat in the Mississippi River Delta. Because 2013's breeding season had gone well and no big hurricanes had ripped up the marsh vegetation that ducks eat, hunters had been optimistic at the start of the season. "We used to be able to predict good and bad years," Weber says.
According to Reynolds, bad years are becoming more normal. Looking back at LDWF survey data from the past two decades, he says "at least 10 percent fewer ducks and geese of all species winter in Louisiana now than they did 15 to 20 years ago." Some birds, including mallards and snow geese (above), have dropped to half their former numbers. Because populations in the flyway as a whole have remained stable, Reynolds suspects more birds are now wintering to the north.
He cites several variables that may be contributing to the shift. Climate change, for example, has resulted in warmer winter temperatures and later freezes in northern states. In addition, new rice varieties are allowing farmers to grow the crop farther north than they once could, and some waterfowl species thrive in flooded rice fields.
Coastal Louisiana suffers the highest rate of land loss in the nation. Since the 1930s, 1,900 square miles of this biologically rich habitat have been transformed into open water, and today an area the size of a football field is lost every hour. Subsidence and wave erosion contribute to wetlands loss. But humans have greatly exacerbated these natural processes by constructing levees, dams and other structures that prevent the Mississippi River from delivering the soil and nutrients that create new land, as it historically did during annual spring flooding. Thousands of pipelines and canals, meanwhile, speed up the destruction by allowing vegetation-killing salt water into the marshes. "Not only are we destroying our wetlands, we are disrupting the natural cycle that can rebuild them," explains David Muth, director of NWF's Mississippi River Delta Restoration Campaign, which works with partners regionally and nationally to restore the coast by reconnecting the river with its delta.
According to a recent calculation by waterfowl scientists, habitat loss means coastal Louisiana today can support some 3 million fewer ducks than it did in the 1970s. But with overall duck numbers 43 percent above their 1955 to 2013 long-term average, according to a July 2014 U.S. Fish and Wildlife Service report, does it matter whether the birds spend winter feeding in a Louisiana marsh or in a rice field in Missouri?
High-quality winter habitat is crucial for waterfowl to build fat reserves they need to survive long migrations to breeding territories. According to DU, ducks use 12 times more energy in flight than at rest, and a single female mallard typically burns about 1.8 million calories during a one-way trip from southern Louisiana to Saskatchewan. If a bird cannot accumulate enough fat during winter, it may leave late to migrate north. Late arrival on the breeding grounds means a female will have lower nesting success than if she had arrived on time. She may even forgo nesting entirely.
"For a bird to successfully reproduce, it must be fit, and for a duck being fit means being fat," Moorman says. Waterfowl also require fat reserves to survive winter. During the cold months, a mallard, for instance, needs enough stored fat to last five to seven days without eating, Moorman says. "That way, it can wait out a cold snap and still have a fuel tank full enough to go elsewhere if the bad weather doesn't break." As evidence that Gulf Coast habitats are suffering, he cites research that found lower survival rates for wintering female northern pintails (below) in southwest Louisiana and coastal Texas than in other regions the birds overwinter.
Fortunately, a large-scale effort to restore Louisiana's disappearing wetlands is closer to reality today than ever before—and Lambert's duck-hunting lease near Buras provides an ideal example of how such an undertaking would work. Located about midway between New Orleans and the state's iconic "bird's foot" at the tip of the delta, Lambert's office and lodge lie on a narrow strip of land bisected by Highway 23. On the west side of the highway, where the Mississippi River is hemmed in by levees, there is nothing but open water, all of it formerly wetlands.
For more than three decades, the threat coastal land loss poses to wildlife and human lives has been well recognized, particularly among people like Lambert who live on the front lines. Yet attempts to tackle the problem so far have been small, fragmented and poorly funded. "When you're losing 16 square miles a year, a half million dollars here and a half million there doesn't make much of a difference," Muth says.
But these days, Muth and his colleagues are feeling more optimistic. Four years ago, when the Deepwater Horizon disaster spewed some 185 million gallons of oil into the Gulf of Mexico, national concern about the spill also focused attention on the long-term crisis of coastal land loss. Now, as legal processes move forward to require responsible parties to pay for the oil damage, a large portion of both civil and criminal fines are being directed toward restoration. Particularly encouraging, the RESTORE Act—passed in 2012 after pressure from NWF, its conservation partners and hunters and anglers nationwide—requires 80 percent of U.S. Clean Water Act penalties go to restoring the Gulf Coast, an amount that could total between $4 billion and $20 billion. Already, Louisiana has received $40 million from BP criminal fines to begin constructing what will be the first large-scale, river-diversion project at Myrtle Grove, 35 miles upriver from Buras.
As restoration moves closer to reality, however, opposition is emerging, primarily from those who harvest oysters, speckled trout and some kinds of shrimp, species whose distributions may change as freshwater inflows alter coastal salinity levels. Rather than diverting river water into the system, opponents favor building new land by dredging sand from the bottom. "But without the ongoing input of new sediment, marshes built by dredge, like new cars, begin to depreciate as soon as you get them," Muth says. He points out that the Mississippi—nicknamed the Big Muddy—carries four times more sediment in suspension than what falls to the bottom. "The river's suspended sediment, the mud that built the delta, cannot be captured by dredges," Muth says.
As a member of the Mississippi River Delta Restoration Campaign, funded by the Walton Family Foundation, NWF's South Central Regional Center, Gulf state affiliates and Washington, D.C., office have worked for several years with national and regional partners to restore wetlands that are vital not only for waterfowl but for countless other wildlife species as well as human communities.
In another collaboration, Vanishing Paradise, NWF and its partners are working with thousands of sportsmen and women, hunting and fishing businesses and other organizations to restore Louisiana's waterfowl and fishing habitat by reconnecting the Mississippi River with its wetlands.
A Vanishing Paradise partner and NWF affiliate, Louisiana Wildlife Federation staff and volunteers have played a key role giving voice to the state's sportsmen by gathering thousands of names of supporters, recruiting 168 businesses and generating more than 100 radio or TV interviews and news articles on the need for coastal restoration. Staff and board members also attend public hearings and work with state and federal agency officials on this critical issue. In addition, LWF has created an army of nearly 7,000 online sportsmen activists supporting coastal restoration through the Louisiana Camo Coalition.
Senior Editor Laura Tangley visited NWF staff and partners in Louisiana to report for this story.


Those amazing women at The Dating Divas have put together a one-of-a kind debut event to strengthen marriages — The Dating Divas "Successful Marriages" Tele-Retreat! Best of all — it's FREE! — and you can participate right from the comfort of your own HOME.
Register today by clicking here! You'll find all kinds of great information and see all the fabulous presenters and their fun tele-seminar topics.
Over 1000 participants have already registered so don't delay!
Click here for the tele-seminar presentation slides!
I had the opportunity to sit in on this "tele retreat" early this year and found that it featured a lot of wonderful insights on working on and committing to living in a happy marriage. And best of all, the information was offered for free! I particularly liked the portion on spicing up our marriage. My husband and I tried several new things and I'm glad to say our marriage has never felt stronger. It's like we're newlyweds again. I haven't had a chance to catch Laura's television show, but I have picked up a copy of her book. I've read it twice now and have marked it up tremendously. I hope you present a similar retreat sometime next year. This was such a great help and has made my husband and I a happier couple.


He feels his palm on fire while unpacking the pizza.
So much for Monica getting away with much. As the movie proceeded Dane decided to get up and carefully walk away. Reaching the classroom door he took his exit to let the movie consume his students. As soon as the door sealed murmuring began.
Bree was blown away. Had she done this before. Rounding the corner of the aisle unaware she marched directly into a large fellow. A white male of 6'4, 350 pounds of fluff.


Healthy eating and consistent exercise is the key to success with your health and wellness. But what do you do when you're looking to lose weight and speed up your metabolic rate, and fast? Here are four simple and quick changes you can make to your everyday routine to kick start burning that unwanted fat. However, you could instead just have a look at somewhere like https://www.ice-aesthetic.co.uk and make an appointment for cryolipolysis treatment that could help you get rid of some of that nasty fat tissue.
As we all know, breakfast is the most important meal of the day. It energizes you and gets your metabolism going. If your breakfast needs aren't met- your day starts off on the wrong foot. Individuals who skip breakfast tend to be cranky, tired and are more likely to be obese. According to a study done by Wolfson Medical Center and the Hebrew University of Jerusalem, women who ate their largest meals of the day at breakfast were more likely to lose weight and decrease waistline circumference than those who ate their largest meal at dinner. So make sure to fit in a hearty protein filled breakfast everyday.
Pro Tip: Try eating a plate of eggs for breakfast. This has proven to leave you feeling full longer throughout the day.
Iron plays an important role in your metabolic process; Iron helps to produce hemoglobin which is the substance that carries oxygen to your muscles to burn fat. If your diet doesn't include enough iron, you may be at risk for developing an iron deficiency anemia. Eating lean meats, beans and spinach is a great way to consume your daily dose of iron-rich foods. A majority of people's consumption of iron comes from meats, so vegetarians have to be mindful and make sure to place iron rich foods into their diets.
Pro Tip: Throw some spinach in your lunch salad or sandwich and boost your iron intake, without making a huge dietary change.
The caffeine in your coffee or tea is helping you to increase your metabolic rate. Caffeine kick starts the nervous system by sending signals to fat cells and telling them to break down the fat. In fact, studies have shown that caffeine can increase your metabolism anywhere from 3 to 11 percent.
Pro Tip: Drink one glass of green tea a day on top of your coffee. It's recommended that you only consume up to 400mg of caffeine per day, so make sure to track your intake.
Eating spicy foods can kickstart your fat burning. Studies have shown that consumption of hot peppers or very spicy foods can increase your metabolic rate by 20% for about 30 minutes. This is because the capsaicin, which is the main element in chilis, creates heat generation and raises body temperature. Spicy foods aren't for everyone. If you're someone who isn't fond of the spice, start off slow and work your way to a flaming hot diet.
Pro Tip: Keep a container of red-pepper flakes at your disposal to throw on all of your meals.
We've shared with you some simple things to incorporate into your diet to get that fat burning in no time. These tips combined with a healthy portion controlled diet and at least 150 minutes of exercise each week will help you to feel healthier in no time.
What are some things you incorporate into your everyday routine, to help you stay healthy and burn excess fat? Share in the comments below.


My friend, Sheldon Richman, wrote this essay in 1999 for the Cato Institute. Nothing seems to change but the political party in charge of the hubris.
Unfortunately, the fresh example of the Iran-Iraq War has not deterred either the policymakers or their expert allies in the private sector. As if their support for Iraq had been a resounding success, they embraced Syria's Hafez Assad and Iran in the conflict with Iraq, blind to what effects that may have in coming years.
Saddam Hussein's invasion of Kuwait on August 2, 1990, underscored more than one irony of prior U.S. policy. U.S. aid to Saddam during his eight-year war with Iran is only one of those ironies.(216) Another is that although President Bush emphatically rejected Saddam's attempt to link the invasion to the plight of the Palestinians, Bush may yet face enormous Arab pressure to address that problem.
Bush offered several reasons for his response to Saddam's actions, a response that included the cobbling of an international coalition of nations. The initial military deployment was to deter an Iraqi attack on Saudi Arabia. Then, ostensibly to drive Iraq from Kuwait, Bush went to the United Nations to have an economic blockade, an act of war, imposed, although American ships were already in place. Vowing to usher in a "new world order," Bush declared that, in the first test of the post-cold-war world, unprovoked aggression and the toppling of a "legitimate" government (read: quasi-feudal monarchy) by a tyrant comparable to Hitler could not be tolerated. The Munich analogy was rolled out more than once. Although American intervention was lightly shrouded in the mantle of the United Nations and collective security, Bush made it clear that no country but the United States could have spearheaded the effort. Bush and other public officials, including Secretary Baker and Senate Minority Leader Robert Dole, raised the less lofty issue of oil and the purported danger to the U.S. economy ("our way of life"), although that argument had been discredited early in the crisis.(217) When the specter of Iraq's controlling 40 percent of the proven oil reserves did not spook the American public, President Bush insisted that the intervention was not about oil but about aggression. He also defended his policy in terms of protecting the Americans held hostage by Saddam Hussein, although they were not taken hostage until after the policy was launched, and of the economic damage being inflicted on the fledgling democracies of Eastern Europe, although the rise in oil prices resulted from Bush's own embargo.
Two days after the November election, the president announced a doubling of the military deployment to provide an "offensive option." Faith in the blockade was abandoned. On Thanksgiving Day 1990 Bush added a new justification of the possible need for war: Saddam's apparent effort to develop nuclear weapons, which, Bush implied, would endanger the American people. The speech followed by days the publication of a New York Times opinion poll in which a majority of respondents had said that a nuclear threat was the one reason they would be willing to support military action against Iraq. Thus, in faithful Orwellian 1984 fashion, the official U.S. attitude toward a recent ally turned 180 degrees.
The hollowness of the Bush administration's reasons, particularly the highly selective stand against aggression, indicates that the president sees the Middle East as his predecessors saw it, as a U.S. sphere of influence in which rival interests may not compete. Saddam's offense did not lie in occupying a neighbor (partners Turkey, Syria, China, and the Soviet Union, as well as Israel, had done that), or in murdering "his own people" (China's leaders and Syria's Hafez Assad had done that), or in having nuclear weapons (several unsavory states have them and more are in the process of acquiring arsenals). Rather, his offense lay in upsetting the status quo in an area where the United States had vowed repeatedly to go to war, if necessary, to prevent adverse change. Bush's policy was a reaffirmation of U.S. claims in the Middle East, in case anyone thought that the end of the cold war made them obsolete. As he put it, the lesson of the war against Iraq is that "what we say goes."(218) Related reasons for the policy include the need for a new mission for a defense establishment threatened by the public's demands for a peace dividend; the desire to test new weapons; and the need to distract the public from troubling domestic issues, such as the exploding budget deficit and higher taxes.
One outcome of U.S. intervention has been immense Arab pressure on the United States to settle the Palestinian question, something that worries Israel. Bush's Arab coalition partners have a strong case when they argue that the United States cannot justify its double standard for Iraq and Israel. Unfortunately, few in the region will argue that Bush should disengage and let the parties solve the problem themselves. At best, Arab pressure may prompt him to change the nuances of U.S. intervention, but it is doubtful Bush will be willing or able to try to change the Shamir government's position on the occupied territories. Israel, for one thing, managed to rehabilitate its public image in the United States by its decision to stay out of the war.
Pax Americana -- we ain't leaving Iraq.


text:
// Copyright (c) 2012, Cornell University
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//     * Redistributions of source code must retain the above copyright notice,
//       this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of HyperDex nor the names of its contributors may be
//       used to endorse or promote products derived from this software without
//       specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.

#define __STDC_LIMIT_MACROS

// C
#include <cstring>
#include <stdint.h>

// STL
#include <algorithm>
#include <tr1/functional>

// e
#include <e/endian.h>
#include <e/safe_math.h>

// HyperDaemon
#include "hyperdex.h"
#include "hyperdaemon/datatypes.h"

// iostream
#include <iostream>

// The below functions often make use of a call to "validate" with the following
// prototype:
//
//     bool (*validate_TYPE)(const uint8_t** ptr, uint32_t* sz, const uint8_t* end)
//
// This returns true if it is able to decode a serialized object of type TYPE
// starting at the memory pointed to by *ptr.  It will never read at values >=
// end.  The 32-bit value "sz" is provided for use with other *_TYPE functions,
// and has a meaning dependent upon TYPE.  The value should be preserved for
// future calls into other *_TYPE functions, but it should not be interpreted
// for any meaning.
//
// On success, *ptr will be changed to point to one byte past the end of the
// structure decoded.

template <typename T>
inline int
_cmp(T a, T b)
{
    if (a < b)
    {
        return -1;
    }
    else if (a > b)
    {
        return 1;
    }

    return 0;
}

static bool
_sort_comparator(int (*compare_micros)(const hyperdex::microop* lhs, const hyperdex::microop* rhs),
                const hyperdex::microop& lhs, const hyperdex::microop& rhs)
{
    return compare_micros(&lhs, &rhs) < 0;
}

static void
_sort_ops(hyperdex::microop* begin, hyperdex::microop* end,
          int (*compare_micros)(const hyperdex::microop* lhs, const hyperdex::microop* rhs))
{
    std::sort(begin, end,
              std::tr1::bind(_sort_comparator, compare_micros,
                             std::tr1::placeholders::_1,
                             std::tr1::placeholders::_2));
}

/////////////////////////////// String Primitives //////////////////////////////

static int
compare_string(const uint8_t* a_ptr, uint32_t a_sz,
               const uint8_t* b_ptr, uint32_t b_sz)
{
    int cmp = memcmp(a_ptr + sizeof(uint32_t),
                     b_ptr + sizeof(uint32_t),
                     std::min(a_sz, b_sz));

    if (cmp == 0)
    {
        return _cmp(a_sz, b_sz);
    }

    return cmp;
}

static int
compare_string_micros_arg1(const hyperdex::microop* lhs,
                           const hyperdex::microop* rhs)
{
    int cmp = memcmp(lhs->arg1.data(),
                     rhs->arg1.data(),
                     std::min(lhs->arg1.size(), rhs->arg1.size()));

    if (cmp == 0)
    {
        return _cmp(lhs->arg1.size(), rhs->arg1.size());
    }

    return cmp;
}

static int
compare_string_micros_arg2(const hyperdex::microop* lhs,
                           const hyperdex::microop* rhs)
{
    int cmp = memcmp(lhs->arg2.data(),
                     rhs->arg2.data(),
                     std::min(lhs->arg2.size(), rhs->arg2.size()));

    if (cmp == 0)
    {
        return _cmp(lhs->arg2.size(), rhs->arg2.size());
    }

    return cmp;
}

static int
compare_string_micro_arg1(const uint8_t* ptr, uint32_t ptr_sz,
                          const hyperdex::microop* op)
{
    int cmp = memcmp(ptr + sizeof(uint32_t),
                     op->arg1.data(),
                     std::min(static_cast<size_t>(ptr_sz), op->arg1.size()));

    if (cmp == 0)
    {
        return _cmp(static_cast<size_t>(ptr_sz), op->arg1.size());
    }

    return cmp;
}

static int
compare_string_micro_arg2(const uint8_t* ptr, uint32_t ptr_sz,
                          const hyperdex::microop* op)
{
    int cmp = memcmp(ptr + sizeof(uint32_t),
                     op->arg2.data(),
                     std::min(static_cast<size_t>(ptr_sz), op->arg2.size()));

    if (cmp == 0)
    {
        return _cmp(static_cast<size_t>(ptr_sz), op->arg2.size());
    }

    return cmp;
}

static uint8_t*
copy_string_from_serialized(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz)
{
    memmove(writeto, ptr, ptr_sz + sizeof(uint32_t));
    return writeto + ptr_sz + sizeof(uint32_t);
}

static uint8_t*
copy_string_from_micro_arg1(uint8_t* writeto, const hyperdex::microop* op)
{
    writeto = e::pack32le(op->arg1.size(), writeto);
    memmove(writeto, op->arg1.data(), op->arg1.size());
    return writeto + op->arg1.size();
}

static uint8_t*
copy_string_from_micro_arg2(uint8_t* writeto, const hyperdex::microop* op)
{
    writeto = e::pack32le(op->arg2.size(), writeto);
    memmove(writeto, op->arg2.data(), op->arg2.size());
    return writeto + op->arg2.size();
}

static e::slice
string_slice_from_serialized(const uint8_t* ptr, uint32_t ptr_sz, std::vector<uint8_t>*)
{
    return e::slice(ptr + sizeof(uint32_t), ptr_sz);
}

static bool
validate_string(const uint8_t** ptr, uint32_t* sz, const uint8_t* end)
{
    if (static_cast<size_t>(end - *ptr) < sizeof(uint32_t))
    {
        return false;
    }

    *ptr = e::unpack32le(*ptr, sz);
    *ptr += *sz;
    return *ptr <= end;
}

static bool
validate_string_micro(const hyperdex::microop* /*op*/)
{
    return true;
}

static uint8_t*
apply_string(const e::slice& old_value,
             const hyperdex::microop* ops,
             size_t num_ops,
             uint8_t* writeto,
             hyperdex::network_returncode* error)
{
    using namespace hyperdex;

    e::slice base = old_value;
    bool set_base = false;
    const hyperdex::microop* prepend = NULL;
    const hyperdex::microop* append = NULL;

    for (size_t i = 0; i < num_ops; ++i)
    {
        switch (ops[i].action)
        {
            case OP_SET:

                if (set_base || prepend || append)
                {
                    std::cout<<"flag 1"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
                }

                base = ops[i].arg1;
                set_base = true;
                break;
            case OP_STRING_PREPEND:

                if (set_base || prepend)
                {
                    std::cout<<"flag 2"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
                }

                prepend = ops + i;
                break;
            case OP_STRING_APPEND:

                if (set_base || append)
                {
                    std::cout<<"flag 3"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
                }

                append = ops + i;
                break;
            case OP_INT64_ADD:
            case OP_INT64_SUB:
            case OP_INT64_MUL:
            case OP_INT64_DIV:
            case OP_INT64_MOD:
            case OP_INT64_AND:
            case OP_INT64_OR:
            case OP_INT64_XOR:
            case OP_LIST_LPUSH:
            case OP_LIST_RPUSH:
            case OP_SET_ADD:
            case OP_SET_REMOVE:
            case OP_SET_INTERSECT:
            case OP_SET_UNION:
            case OP_MAP_ADD:
            case OP_MAP_REMOVE:
            case OP_FAIL:
            default:
                std::cout<<"flag 4"<<std::endl;
                *error = NET_BADMICROS;
                return NULL;
        }
    }

    if (prepend)
    {
        memmove(writeto, prepend->arg1.data(), prepend->arg1.size());
        writeto += prepend->arg1.size();
    }

    memmove(writeto, base.data(), base.size());
    writeto += base.size();

    if (append)
    {
        memmove(writeto, append->arg1.data(), append->arg1.size());
        writeto += append->arg1.size();
    }

    return writeto;
}

/////////////////////////////// Int64 Primitives ///////////////////////////////

static int
compare_int64(const uint8_t* a_ptr, uint32_t a_sz,
              const uint8_t* b_ptr, uint32_t b_sz)
{
    assert(a_sz == sizeof(int64_t));
    assert(b_sz == sizeof(int64_t));
    int64_t a;
    int64_t b;
    e::unpack64le(a_ptr, &a);
    e::unpack64le(b_ptr, &b);
    return _cmp(a, b);
}

static int
compare_int64_micros_arg1(const hyperdex::microop* lhs,
                          const hyperdex::microop* rhs)
{
    int64_t lhsnum = 0;
    int64_t rhsnum = 0;
    e::unpack64le(lhs->arg1.data(), &lhsnum);
    e::unpack64le(rhs->arg1.data(), &rhsnum);
    return _cmp(lhsnum, rhsnum);
}

static int
compare_int64_micros_arg2(const hyperdex::microop* lhs,
                          const hyperdex::microop* rhs)
{
    int64_t lhsnum = 0;
    int64_t rhsnum = 0;
    e::unpack64le(lhs->arg2.data(), &lhsnum);
    e::unpack64le(rhs->arg2.data(), &rhsnum);
    return _cmp(lhsnum, rhsnum);
}

static int
compare_int64_micro_arg1(const uint8_t* ptr, uint32_t ptr_sz,
                         const hyperdex::microop* op)
{
    assert(ptr_sz == sizeof(int64_t));
    int64_t lhsnum = 0;
    int64_t rhsnum = 0;
    e::unpack64le(ptr, &lhsnum);
    e::unpack64le(op->arg1.data(), &rhsnum);
    return _cmp(lhsnum, rhsnum);
}

static int
compare_int64_micro_arg2(const uint8_t* ptr, uint32_t ptr_sz,
                         const hyperdex::microop* op)
{
    assert(ptr_sz == sizeof(int64_t));
    int64_t lhsnum = 0;
    int64_t rhsnum = 0;
    e::unpack64le(ptr, &lhsnum);
    e::unpack64le(op->arg2.data(), &rhsnum);
    return _cmp(lhsnum, rhsnum);
}

static uint8_t*
copy_int64_from_serialized(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz)
{
    assert(ptr_sz == sizeof(int64_t));
    memmove(writeto, ptr, sizeof(int64_t));
    return writeto + sizeof(int64_t);
}

static uint8_t*
copy_int64_from_micro_arg1(uint8_t* writeto, const hyperdex::microop* op)
{
    assert(op->arg1.size() == sizeof(int64_t));
    memmove(writeto, op->arg1.data(), sizeof(int64_t));
    return writeto + sizeof(int64_t);
}

static uint8_t*
copy_int64_from_micro_arg2(uint8_t* writeto, const hyperdex::microop* op)
{
    assert(op->arg2.size() == sizeof(int64_t));
    memmove(writeto, op->arg2.data(), sizeof(int64_t));
    return writeto + sizeof(int64_t);
}

static e::slice
int64_slice_from_serialized(const uint8_t* ptr, uint32_t ptr_sz, std::vector<uint8_t>*)
{
    return e::slice(ptr, ptr_sz);
}

static bool
validate_int64(const uint8_t** ptr, uint32_t* sz, const uint8_t* end)
{
    uint32_t tmp = end - *ptr;

    if (tmp < sizeof(int64_t))
    {
        return false;
    }

    *sz = sizeof(int64_t);
    *ptr += sizeof(int64_t);
    return true;
}

static bool
validate_int64_micro_arg1(const hyperdex::microop* op)
{
    return op->arg1.size() == sizeof(int64_t);
}

static bool
validate_int64_micro_arg2(const hyperdex::microop* op)
{
    return op->arg2.size() == sizeof(int64_t);
}

static uint8_t*
apply_int64(const e::slice& old_value,
            const hyperdex::microop* ops,
            size_t num_ops,
            uint8_t* writeto,
            hyperdex::network_returncode* error)
{
    using namespace hyperdex;

    int64_t number = 0;

    if (old_value.size())
    {
        e::unpack64le(old_value.data(), &number);
    }

    for (size_t i = 0; i < num_ops; ++i)
    {
        const hyperdex::microop* op = ops + i;

        if (!validate_int64_micro_arg1(op))
        {
            std::cout<<"flag 5"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int64_t arg;
        e::unpack64le(op->arg1.data(), &arg);

        switch (op->action)
        {
            case OP_SET:
                number = arg;
                break;
            case OP_INT64_ADD:
                if (!e::safe_add(number, arg, &number))
                {
                    *error = hyperdex::NET_OVERFLOW;
                    return NULL;
                }
                break;
            case OP_INT64_SUB:
                if (!e::safe_sub(number, arg, &number))
                {
                    *error = hyperdex::NET_OVERFLOW;
                    return NULL;
                }
                break;
            case OP_INT64_MUL:
                if (!e::safe_mul(number, arg, &number))
                {
                    *error = hyperdex::NET_OVERFLOW;
                    return NULL;
                }
                break;
            case OP_INT64_DIV:
                if (!e::safe_div(number, arg, &number))
                {
                    *error = hyperdex::NET_OVERFLOW;
                    return NULL;
                }
                break;
            case OP_INT64_MOD:
                if (!e::safe_mod(number, arg, &number))
                {
                    *error = hyperdex::NET_OVERFLOW;
                    return NULL;
                }
                break;
            case OP_INT64_AND:
                number &= arg;
                break;
            case OP_INT64_OR:
                number |= arg;
                break;
            case OP_INT64_XOR:
                number ^= arg;
                break;
            case OP_STRING_APPEND:
            case OP_STRING_PREPEND:
            case OP_LIST_LPUSH:
            case OP_LIST_RPUSH:
            case OP_SET_ADD:
            case OP_SET_REMOVE:
            case OP_SET_INTERSECT:
            case OP_SET_UNION:
            case OP_MAP_ADD:
            case OP_MAP_REMOVE:
            case OP_FAIL:
            default:
                std::cout<<"flag 6"<<std::endl;
                *error = NET_BADMICROS;
                return NULL;
        }
    }

    return e::pack64le(number, writeto);
}

//////////////////////////// Abstract List Functions ///////////////////////////

static bool
validate_list(bool (*validate_elem)(const uint8_t** ptr,
                                    uint32_t* ptr_sz,
                                    const uint8_t* end),
              const e::slice& list)
{
    const uint8_t* ptr = list.data();
    const uint8_t* const end = list.data() + list.size();

    while (ptr < end)
    {
        uint32_t sz;

        if (!validate_elem(&ptr, &sz, end))
        {
            return false;
        }
    }

    return ptr == end;
}

static uint8_t*
apply_list(bool (*validate_elem)(const uint8_t** ptr,
                                 uint32_t* ptr_sz,
                                 const uint8_t* end),
           bool (*validate_elem_micro)(const hyperdex::microop* op),
           uint8_t* (*copy_elem_from_micro)(uint8_t* writeto,
                                            const hyperdex::microop* op),
           const e::slice& old_value,
           const hyperdex::microop* ops,
           size_t num_ops,
           uint8_t* writeto,
           hyperdex::network_returncode* error)
{
    if (num_ops == 1 && ops->action == hyperdex::OP_SET)
    {
        if (!validate_list(validate_elem, ops->arg1))
        {
            std::cout<<"flag 7"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }

        memmove(writeto, ops->arg1.data(), ops->arg1.size());
        return writeto + ops->arg1.size();
    }

    // Apply the lpush ops
    for (ssize_t i = num_ops - 1; i >= 0; --i)
    {
        if (ops[i].action != hyperdex::OP_LIST_LPUSH)
        {
            continue;
        }

        if (!validate_elem_micro(ops + i))
        {
            std::cout<<"flag 8"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }

        writeto = copy_elem_from_micro(writeto, ops + i);
    }

    // Copy the middle of the list
    memmove(writeto, old_value.data(), old_value.size());
    writeto += old_value.size();

    // Apply the rpush ops
    for (size_t i = 0; i < num_ops; ++i)
    {
        if (ops[i].action != hyperdex::OP_LIST_RPUSH)
        {
            if (ops[i].action != hyperdex::OP_LIST_LPUSH)
            {
                std::cout<<"flag 8"<<std::endl;
                *error = hyperdex::NET_BADMICROS;
                return NULL;
            }

            continue;
        }

        if (!validate_elem_micro(ops + i))
        {
            std::cout<<"flag 9"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }

        writeto = copy_elem_from_micro(writeto, ops + i);
    }

    return writeto;
}

//////////////////////////// Abstract Set Functions ////////////////////////////

static bool
validate_set(bool (*validate_elem)(const uint8_t** ptr,
                                   uint32_t* ptr_sz,
                                   const uint8_t* end),
             int (*compare_elem)(const uint8_t* a_ptr, uint32_t a_sz,
                                 const uint8_t* b_ptr, uint32_t b_sz),
             const e::slice& set)
{
    bool has_old = false;
    uint32_t old_sz = 0;
    const uint8_t* old_ptr = NULL;
    const uint8_t* ptr = set.data();
    const uint8_t* const end = set.data() + set.size();

    while (ptr < end)
    {
        uint32_t sz;
        const uint8_t* next_ptr = ptr;

        if (!validate_elem(&next_ptr, &sz, end))
        {
            return false;
        }

        // sz, ptr are size and pointer to the element we just validated.
        // old_sz, old_ptr are size and pointer to the previous element
        // next_ptr is the start of the next element

        if (has_old)
        {
            int cmp = compare_elem(old_ptr, old_sz, ptr, sz);

            if (cmp >= 0)
            {
                return false;
            }
        }

        old_sz = sz;
        old_ptr = ptr;
        has_old = true;
        ptr = next_ptr;
    }

    return ptr == end;
}

static uint8_t*
apply_set_add_remove(bool (*validate_elem)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                     bool (*validate_elem_micro)(const hyperdex::microop* op),
                     int (*compare_micros)(const hyperdex::microop* lhs, const hyperdex::microop* rhs),
                     int (*compare_elem)(const uint8_t* ptr, uint32_t ptr_sz, const hyperdex::microop* op),
                     uint8_t* (*copy_elem_from_set)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                     uint8_t* (*copy_elem_from_micro)(uint8_t* writeto, const hyperdex::microop* op),
                     const e::slice& old_value,
                     const hyperdex::microop* ops,
                     size_t num_ops,
                     uint8_t* writeto,
                     hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(num_ops > 0);

    if (!validate_elem_micro(ops))
    {
        std::cout<<"flag 10"<<std::endl;
        *error = hyperdex::NET_BADMICROS;
        return NULL;
    }

    // Verify sorted order of the microops.
    for (size_t i = 0; i < num_ops - 1; ++i)
    {
        if (!validate_elem_micro(ops + i + 1))
        {
            std::cout<<"flag 11"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }

        if (compare_micros(ops + i, ops + i + 1) >= 0)
        {
            std::cout<<"flag 12"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }
    }

    const uint8_t* ptr = old_value.data();
    const uint8_t* const end = old_value.data() + old_value.size();
    size_t i = 0;

    while (ptr < end && i < num_ops)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* next_from_set = ptr;

        if (!validate_elem(&next_from_set, &ptr_sz, end))
        {
            std::cout<<"flag 13"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int cmp = compare_elem(ptr, ptr_sz, ops + i);

        if (cmp < 0)
        {
            writeto = copy_elem_from_set(writeto, ptr, ptr_sz);
            ptr = next_from_set;
        }
        else if (cmp > 0)
        {
            switch (ops[i].action)
            {
                case OP_SET_ADD:
                    writeto = copy_elem_from_micro(writeto, ops + i);
                    ++i;
                    break;
                case OP_SET_REMOVE:
                    // Nothing to remove
                    ++i;
                    break;
                case OP_SET_INTERSECT:
                case OP_SET_UNION:
                case OP_STRING_APPEND:
                case OP_STRING_PREPEND:
                case OP_INT64_ADD:
                case OP_INT64_SUB:
                case OP_INT64_MUL:
                case OP_INT64_DIV:
                case OP_INT64_MOD:
                case OP_INT64_AND:
                case OP_INT64_OR:
                case OP_INT64_XOR:
                case OP_LIST_LPUSH:
                case OP_LIST_RPUSH:
                case OP_MAP_ADD:
                case OP_MAP_REMOVE:
                case OP_SET:
                case OP_FAIL:
                default:
                    std::cout<<"flag 14"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
            }
        }
        else
        {
            switch (ops[i].action)
            {
                case OP_SET_ADD:
                    writeto = copy_elem_from_micro(writeto, ops + i);
                    ptr = next_from_set;
                    ++i;
                    break;
                case OP_SET_REMOVE:
                    ptr = next_from_set;
                    ++i;
                    break;
                case OP_SET_INTERSECT:
                case OP_SET_UNION:
                case OP_STRING_APPEND:
                case OP_STRING_PREPEND:
                case OP_INT64_ADD:
                case OP_INT64_SUB:
                case OP_INT64_MUL:
                case OP_INT64_DIV:
                case OP_INT64_MOD:
                case OP_INT64_AND:
                case OP_INT64_OR:
                case OP_INT64_XOR:
                case OP_LIST_LPUSH:
                case OP_LIST_RPUSH:
                case OP_MAP_ADD:
                case OP_MAP_REMOVE:
                case OP_SET:
                case OP_FAIL:
                default:
                    std::cout<<"flag 15"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
            }
        }
    }

    while (ptr < end)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* next_from_set = ptr;

        if (!validate_elem(&next_from_set, &ptr_sz, end))
        {
            std::cout<<"flag 16"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        writeto = copy_elem_from_set(writeto, ptr, ptr_sz);
        ptr = next_from_set;
    }

    while (i < num_ops)
    {
        switch (ops[i].action)
        {
            case OP_SET_ADD:
                writeto = copy_elem_from_micro(writeto, ops + i);
                ++i;
                break;
            case OP_SET_REMOVE:
                ++i;
                break;
            case OP_SET_INTERSECT:
            case OP_SET_UNION:
            case OP_STRING_APPEND:
            case OP_STRING_PREPEND:
            case OP_INT64_ADD:
            case OP_INT64_SUB:
            case OP_INT64_MUL:
            case OP_INT64_DIV:
            case OP_INT64_MOD:
            case OP_INT64_AND:
            case OP_INT64_OR:
            case OP_INT64_XOR:
            case OP_LIST_LPUSH:
            case OP_LIST_RPUSH:
            case OP_MAP_ADD:
            case OP_MAP_REMOVE:
            case OP_SET:
            case OP_FAIL:
            default:
                std::cout<<"flag 17"<<std::endl;
                *error = NET_BADMICROS;
                return NULL;
        }
    }

    return writeto;
}

static uint8_t*
apply_set_set(bool (*validate_set)(const e::slice& data),
              const hyperdex::microop* op,
              uint8_t* writeto,
              hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(op->action == hyperdex::OP_SET);

    if (!validate_set(op->arg1))
    {
        std::cout<<"flag 18"<<std::endl;
        *error = NET_BADMICROS;
        return NULL;
    }

    memmove(writeto, op->arg1.data(), op->arg1.size());
    return writeto + op->arg1.size();
}

static uint8_t*
apply_set_intersect(bool (*validate_set)(const e::slice& data),
                    bool (*validate_elem)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                    int (*compare_elem)(const uint8_t* a_ptr, uint32_t a_sz,
                                        const uint8_t* b_ptr, uint32_t b_sz),
                    uint8_t* (*copy_elem_from_set)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                    const e::slice& old_value,
                    const hyperdex::microop* op,
                    uint8_t* writeto,
                    hyperdex::network_returncode* error)
{
    // There's an efficiency tradeoff in this function.  We could validate that
    // the user-provided set is sorted, and then run a merge algorithm, or we
    // could validate that it's sorted on the fly.  It's easier to pre-validate
    // that the set is sorted, and the efficiency trade-off is on the order of
    // constants.
    using namespace hyperdex;
    assert(op->action == hyperdex::OP_SET_INTERSECT);

    if (!validate_set(op->arg1))
    {
        std::cout<<"flag 19"<<std::endl;
        *error = NET_BADMICROS;
        return NULL;
    }

    const uint8_t* set_ptr = old_value.data();
    const uint8_t* const set_end = old_value.data() + old_value.size();
    const uint8_t* new_ptr = op->arg1.data();
    const uint8_t* const new_end = op->arg1.data() + op->arg1.size();

    // We only need to consider this case.
    while (set_ptr < set_end && new_ptr < new_end)
    {
        uint32_t set_sz = 0;
        uint32_t new_sz = 0;
        const uint8_t* next_set_ptr = set_ptr;
        const uint8_t* next_new_ptr = new_ptr;

        if (!validate_elem(&next_set_ptr, &set_sz, set_end))
        {
            std::cout<<"flag 20"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        if (!validate_elem(&next_new_ptr, &new_sz, new_end))
        {
            std::cout<<"flag 21"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int cmp = compare_elem(set_ptr, set_sz, new_ptr, new_sz);

        if (cmp < 0)
        {
            set_ptr = next_set_ptr;
        }
        else if (cmp > 0)
        {
            new_ptr = next_new_ptr;
        }
        else
        {
            writeto = copy_elem_from_set(writeto, set_ptr, new_sz);
            set_ptr = next_set_ptr;
            new_ptr = next_new_ptr;
        }
    }

    return writeto;
}

static uint8_t*
apply_set_union(bool (*validate_set)(const e::slice& data),
                bool (*validate_elem)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                int (*compare_elem)(const uint8_t* a_ptr, uint32_t a_sz,
                                    const uint8_t* b_ptr, uint32_t b_sz),
                uint8_t* (*copy_elem_from_set)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                const e::slice& old_value,
                const hyperdex::microop* op,
                uint8_t* writeto,
                hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(op->action == hyperdex::OP_SET_UNION);

    if (!validate_set(op->arg1))
    {
        std::cout<<"flag 21"<<std::endl;
        *error = NET_BADMICROS;
        return NULL;
    }

    const uint8_t* set_ptr = old_value.data();
    const uint8_t* const set_end = old_value.data() + old_value.size();
    const uint8_t* new_ptr = op->arg1.data();
    const uint8_t* const new_end = op->arg1.data() + op->arg1.size();

    while (set_ptr < set_end && new_ptr < new_end)
    {
        uint32_t set_sz = 0;
        uint32_t new_sz = 0;
        const uint8_t* next_set_ptr = set_ptr;
        const uint8_t* next_new_ptr = new_ptr;

        if (!validate_elem(&next_set_ptr, &set_sz, set_end))
        {
            std::cout<<"flag 22"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        if (!validate_elem(&next_new_ptr, &new_sz, new_end))
        {
            std::cout<<"flag 23"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int cmp = compare_elem(set_ptr, set_sz, new_ptr, new_sz);

        if (cmp < 0)
        {
            writeto = copy_elem_from_set(writeto, set_ptr, set_sz);
            set_ptr = next_set_ptr;
        }
        else if (cmp > 0)
        {
            writeto = copy_elem_from_set(writeto, new_ptr, new_sz);
            new_ptr = next_new_ptr;
        }
        else
        {
            writeto = copy_elem_from_set(writeto, set_ptr, set_sz);
            set_ptr = next_set_ptr;
            new_ptr = next_new_ptr;
        }
    }

    while (set_ptr < set_end)
    {
        uint32_t set_sz = 0;
        const uint8_t* next_set_ptr = set_ptr;

        if (!validate_elem(&next_set_ptr, &set_sz, set_end))
        {
            std::cout<<"flag 24"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        writeto = copy_elem_from_set(writeto, set_ptr, set_sz);
        set_ptr = next_set_ptr;
    }

    while (new_ptr < new_end)
    {
        uint32_t new_sz = 0;
        const uint8_t* next_new_ptr = new_ptr;

        if (!validate_elem(&next_new_ptr, &new_sz, new_end))
        {
            std::cout<<"flag 25"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        writeto = copy_elem_from_set(writeto, new_ptr, new_sz);
        new_ptr = next_new_ptr;
    }

    return writeto;
}

//////////////////////////// Abstract Map Functions ////////////////////////////

static bool
validate_map(bool (*validate_key)(const uint8_t** ptr,
                                  uint32_t* ptr_sz,
                                  const uint8_t* end),
             bool (*validate_value)(const uint8_t** ptr,
                                    uint32_t* ptr_sz,
                                    const uint8_t* end),
             int (*compare_key)(const uint8_t* a_ptr, uint32_t a_sz,
                                const uint8_t* b_ptr, uint32_t b_sz),
             const e::slice& map)
{
    bool has_old = false;
    uint32_t old_sz = 0;
    const uint8_t* old_ptr = NULL;
    const uint8_t* ptr = map.data();
    const uint8_t* const end = map.data() + map.size();
    uint32_t dummy;

    while (ptr < end)
    {
        uint32_t sz;
        const uint8_t* next_ptr = ptr;

        if (!validate_key(&next_ptr, &sz, end))
        {
            return false;
        }

        if (!validate_value(&next_ptr, &dummy, end))
        {
            return false;
        }

        // sz, ptr are size and pointer to the key we just validated.
        // old_sz, old_ptr are size and pointer to the previous key
        // next_ptr is the start of the next key
        // the value associated with the current key is valid

        if (has_old)
        {
            int cmp = compare_key(old_ptr, old_sz, ptr, sz);

            if (cmp >= 0)
            {
                return false;
            }
        }

        old_sz = sz;
        old_ptr = ptr;
        has_old = true;
        ptr = next_ptr;
    }

    return ptr == end;
}

static uint8_t*
apply_map_set(bool (*validate_map)(const e::slice& data),
              const hyperdex::microop* op,
              uint8_t* writeto,
              hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(op->action == hyperdex::OP_SET);

    if (!validate_map(op->arg1))
    {
        std::cout<<"flag 26"<<std::endl;
        *error = NET_BADMICROS;
        return NULL;
    }

    memmove(writeto, op->arg1.data(), op->arg1.size());
    return writeto + op->arg1.size();
}

static uint8_t*
apply_map_add_remove(bool (*validate_key)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                     bool (*validate_val)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                     bool (*validate_elem_micro_arg1)(const hyperdex::microop* op),
                     bool (*validate_elem_micro_arg2)(const hyperdex::microop* op),
                     int (*compare_micros)(const hyperdex::microop* lhs, const hyperdex::microop* rhs),
                     int (*compare_elem)(const uint8_t* ptr, uint32_t ptr_sz, const hyperdex::microop* op),
                     uint8_t* (*copy_key_from_map)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                     uint8_t* (*copy_key_from_micro)(uint8_t* writeto, const hyperdex::microop* op),
                     uint8_t* (*copy_val_from_map)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                     uint8_t* (*copy_val_from_micro)(uint8_t* writeto, const hyperdex::microop* op),
                     const e::slice& old_value,
                     hyperdex::microop* ops,
                     size_t num_ops,
                     uint8_t* writeto,
                     hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(num_ops > 0);

    if ((ops->action != OP_MAP_REMOVE && !validate_elem_micro_arg1(ops)) ||
        !validate_elem_micro_arg2(ops))
    {
        std::cout<<"flag 27"<<std::endl;
        *error = hyperdex::NET_BADMICROS;
        return NULL;
    }

    // Validate and sort the the microops.
    for (size_t i = 0; i < num_ops - 1; ++i)
    {
        if ((ops->action != OP_MAP_REMOVE && !validate_elem_micro_arg1(ops + i + 1)) ||
            !validate_elem_micro_arg2(ops + i + 1))
        {
            std::cout<<"flag 28"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }
    }

    _sort_ops(ops, ops + num_ops, compare_micros);
    const uint8_t* ptr = old_value.data();
    const uint8_t* const end = old_value.data() + old_value.size();
    size_t i = 0;

    while (ptr < end && i < num_ops)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* val_ptr = ptr;

        if (!validate_key(&val_ptr, &ptr_sz, end))
        {
            std::cout<<"flag 29"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        uint32_t val_ptr_sz = 0;
        const uint8_t* next_from_map = val_ptr;

        if (!validate_val(&next_from_map, &val_ptr_sz, end))
        {
            std::cout<<"flag 30"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int cmp = compare_elem(ptr, ptr_sz, ops + i);

        if (cmp < 0)
        {
            writeto = copy_key_from_map(writeto, ptr, ptr_sz);
            writeto = copy_val_from_map(writeto, val_ptr, val_ptr_sz);
            ptr = next_from_map;
        }
        else if (cmp > 0)
        {
            switch (ops[i].action)
            {
                case OP_MAP_ADD:
                    writeto = copy_key_from_micro(writeto, ops + i);
                    writeto = copy_val_from_micro(writeto, ops + i);
                    ++i;
                    break;
                case OP_MAP_REMOVE:
                    // Nothing to remove
                    ++i;
                    break;
                case OP_STRING_APPEND:
                case OP_STRING_PREPEND:
                case OP_INT64_ADD:
                case OP_INT64_SUB:
                case OP_INT64_MUL:
                case OP_INT64_DIV:
                case OP_INT64_MOD:
                case OP_INT64_AND:
                case OP_INT64_OR:
                case OP_INT64_XOR:
                case OP_LIST_LPUSH:
                case OP_LIST_RPUSH:
                case OP_SET_ADD:
                case OP_SET_REMOVE:
                case OP_SET_INTERSECT:
                case OP_SET_UNION:
                case OP_SET:
                case OP_FAIL:
                default:
                    std::cout<<"flag 31"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
            }
        }
        else
        {
            switch (ops[i].action)
            {
                case OP_MAP_ADD:
                    writeto = copy_key_from_micro(writeto, ops + i);
                    writeto = copy_val_from_micro(writeto, ops + i);
                    ptr = next_from_map;
                    ++i;
                    break;
                case OP_MAP_REMOVE:
                    ptr = next_from_map;
                    ++i;
                    break;
                case OP_STRING_APPEND:
                case OP_STRING_PREPEND:
                case OP_INT64_ADD:
                case OP_INT64_SUB:
                case OP_INT64_MUL:
                case OP_INT64_DIV:
                case OP_INT64_MOD:
                case OP_INT64_AND:
                case OP_INT64_OR:
                case OP_INT64_XOR:
                case OP_LIST_LPUSH:
                case OP_LIST_RPUSH:
                case OP_SET_ADD:
                case OP_SET_REMOVE:
                case OP_SET_INTERSECT:
                case OP_SET_UNION:
                case OP_SET:
                case OP_FAIL:
                default:
                    std::cout<<"flag 32"<<std::endl;
                    *error = NET_BADMICROS;
                    return NULL;
            }
        }
    }

    while (ptr < end)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* val_ptr = ptr;

        if (!validate_key(&val_ptr, &ptr_sz, end))
        {
            std::cout<<"flag 33"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        uint32_t val_ptr_sz = 0;
        const uint8_t* next_from_map = val_ptr;

        if (!validate_val(&next_from_map, &val_ptr_sz, end))
        {
            std::cout<<"flag 34"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        writeto = copy_key_from_map(writeto, ptr, ptr_sz);
        writeto = copy_val_from_map(writeto, val_ptr, val_ptr_sz);
        ptr = next_from_map;
    }

    while (i < num_ops)
    {
        switch (ops[i].action)
        {
            case OP_MAP_ADD:
                writeto = copy_key_from_micro(writeto, ops + i);
                writeto = copy_val_from_micro(writeto, ops + i);
                ++i;
                break;
            case OP_MAP_REMOVE:
                ++i;
                break;
            case OP_STRING_APPEND:
            case OP_STRING_PREPEND:
            case OP_INT64_ADD:
            case OP_INT64_SUB:
            case OP_INT64_MUL:
            case OP_INT64_DIV:
            case OP_INT64_MOD:
            case OP_INT64_AND:
            case OP_INT64_OR:
            case OP_INT64_XOR:
            case OP_LIST_LPUSH:
            case OP_LIST_RPUSH:
            case OP_SET_ADD:
            case OP_SET_REMOVE:
            case OP_SET_INTERSECT:
            case OP_SET_UNION:
            case OP_SET:
            case OP_FAIL:
            default:
                std::cout<<"flag 35"<<std::endl;
                *error = NET_BADMICROS;
                return NULL;
        }
    }

    return writeto;
}

static uint8_t*
apply_map_microop(bool (*validate_key)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                  bool (*validate_val)(const uint8_t** ptr, uint32_t* ptr_sz, const uint8_t* end),
                  bool (*validate_elem_micro_arg1)(const hyperdex::microop* op),
                  bool (*validate_elem_micro_arg2)(const hyperdex::microop* op),
                  int (*compare_micros)(const hyperdex::microop* lhs, const hyperdex::microop* rhs),
                  int (*compare_elem)(const uint8_t* ptr, uint32_t ptr_sz, const hyperdex::microop* op),
                  uint8_t* (*copy_key_from_map)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                  uint8_t* (*copy_key_from_micro)(uint8_t* writeto, const hyperdex::microop* op),
                  uint8_t* (*copy_val_from_map)(uint8_t* writeto, const uint8_t* ptr, uint32_t ptr_sz),
                  e::slice (*slice_from_map)(const uint8_t* ptr, uint32_t ptr_sz, std::vector<uint8_t>* tmp),
                  uint8_t* (*apply_pod)(const e::slice& old_value,
                                        const hyperdex::microop* ops,
                                        size_t num_ops,
                                        uint8_t* writeto,
                                        hyperdex::network_returncode* error),
                  const e::slice& old_value,
                  hyperdex::microop* ops,
                  size_t num_ops,
                  uint8_t* writeto,
                  hyperdex::network_returncode* error)
{
    using namespace hyperdex;
    assert(num_ops > 0);

    if (!validate_elem_micro_arg1(ops) ||
        !validate_elem_micro_arg2(ops))
    {
        std::cout<<"flag 36"<<std::endl;
        *error = hyperdex::NET_BADMICROS;
        return NULL;
    }

    // Verify sorted order of the microops.
    for (size_t i = 0; i < num_ops - 1; ++i)
    {
        if (!validate_elem_micro_arg1(ops + i + 1) ||
            !validate_elem_micro_arg2(ops + i + 1))
        {
            std::cout<<"flag 37"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
        }

        // The '=' part of '>=' ensures that we don't need to do anything fancy
        // to accommodate multiple micro ops on the same map key.  We're always
        // guaranteed to be operating on the same object attribute, so we need
        // no special consideration for that.
        //
        // XXX Modify this function to allow multiple micro ops per map key.
        if (compare_micros(ops + i, ops + i + 1) == 0)
        {
            std::cout<<"flag 38"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }
    }

    _sort_ops(ops, ops + num_ops, compare_micros);
    const uint8_t* ptr = old_value.data();
    const uint8_t* const end = old_value.data() + old_value.size();
    size_t i = 0;

    while (ptr < end && i < num_ops)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* val_ptr = ptr;

        if (!validate_key(&val_ptr, &ptr_sz, end))
        {
            std::cout<<"flag 39"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        uint32_t val_ptr_sz = 0;
        const uint8_t* next_from_map = val_ptr;

        if (!validate_val(&next_from_map, &val_ptr_sz, end))
        {
            std::cout<<"flag 40"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        int cmp = compare_elem(ptr, ptr_sz, ops + i);

        if (cmp < 0)
        {
            writeto = copy_key_from_map(writeto, ptr, ptr_sz);
            writeto = copy_val_from_map(writeto, val_ptr, val_ptr_sz);
            ptr = next_from_map;
        }
        else if (cmp > 0)
        {
            e::slice existing("", 0);
            writeto = copy_key_from_micro(writeto, ops + i);
            writeto = apply_pod(existing, ops + i, 1, writeto, error);

            if (!writeto)
            {
                return NULL;
            }

            ++i;
        }
        else
        {
            std::vector<uint8_t> tmp;
            e::slice existing = slice_from_map(val_ptr, val_ptr_sz, &tmp);
            writeto = copy_key_from_micro(writeto, ops + i);
            writeto = apply_pod(existing, ops + i, 1, writeto, error);

            if (!writeto)
            {
                return NULL;
            }

            ptr = next_from_map;
            ++i;
        }
    }

    while (ptr < end)
    {
        uint32_t ptr_sz = 0;
        const uint8_t* val_ptr = ptr;

        if (!validate_key(&val_ptr, &ptr_sz, end))
        {
            std::cout<<"flag 41"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        uint32_t val_ptr_sz = 0;
        const uint8_t* next_from_map = val_ptr;

        if (!validate_val(&next_from_map, &val_ptr_sz, end))
        {
            std::cout<<"flag 42"<<std::endl;
            *error = NET_BADMICROS;
            return NULL;
        }

        writeto = copy_key_from_map(writeto, ptr, ptr_sz);
        writeto = copy_val_from_map(writeto, val_ptr, val_ptr_sz);
        ptr = next_from_map;
    }

    while (i < num_ops)
    {
        e::slice existing("", 0);
        writeto = copy_key_from_micro(writeto, ops + i);
        writeto = apply_pod(existing, ops + i, 1, writeto, error);

        if (!writeto)
        {
            return NULL;
        }

        ++i;
    }

    return writeto;
}

////////////////////////// Instantiated List Functions /////////////////////////

static bool
validate_list_string(const e::slice& list)
{
    return validate_list(validate_string, list);
}

static bool
validate_list_int64(const e::slice& list)
{
    return validate_list(validate_int64, list);
}

static uint8_t*
apply_list_string(const e::slice& old_value,
                  const hyperdex::microop* ops,
                  size_t num_ops,
                  uint8_t* writeto,
                  hyperdex::network_returncode* error)
{
    return apply_list(validate_string,
                      validate_string_micro,
                      copy_string_from_micro_arg1,
                      old_value, ops, num_ops, writeto, error);
}

static uint8_t*
apply_list_int64(const e::slice& old_value,
                 const hyperdex::microop* ops,
                 size_t num_ops,
                 uint8_t* writeto,
                 hyperdex::network_returncode* error)
{
    return apply_list(validate_int64,
                      validate_int64_micro_arg1,
                      copy_int64_from_micro_arg1,
                      old_value, ops, num_ops, writeto, error);
}

////////////////////////// Instantiated Set Functions //////////////////////////

static bool
validate_set_string(const e::slice& set)
{
    return validate_set(validate_string, compare_string, set);
}

static bool
validate_set_int64(const e::slice& set)
{
    return validate_set(validate_int64, compare_int64, set);
}

static uint8_t*
apply_set_string(const e::slice& old_value,
                 const hyperdex::microop* ops,
                 size_t num_ops,
                 uint8_t* writeto,
                 hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_SET_ADD ||
        ops[0].action == hyperdex::OP_SET_REMOVE)
    {
        return apply_set_add_remove(validate_string,
                                    validate_string_micro,
                                    compare_string_micros_arg1,
                                    compare_string_micro_arg1,
                                    copy_string_from_serialized,
                                    copy_string_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_set_set(validate_set_string, ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET_INTERSECT && num_ops == 1)
    {
        return apply_set_intersect(validate_set_string, validate_string,
                                   compare_string, copy_string_from_serialized,
                                   old_value, ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET_UNION && num_ops == 1)
    {
        return apply_set_union(validate_set_string, validate_string,
                               compare_string, copy_string_from_serialized,
                               old_value, ops, writeto, error);
    }
    else
    {
        std::cout<<"flag 43"<<std::endl;
        *error = hyperdex::NET_BADMICROS;
        return NULL;
    }
}

static uint8_t*
apply_set_int64(const e::slice& old_value,
                const hyperdex::microop* ops,
                size_t num_ops,
                uint8_t* writeto,
                hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_SET_ADD ||
        ops[0].action == hyperdex::OP_SET_REMOVE)
    {
        return apply_set_add_remove(validate_int64,
                                    validate_int64_micro_arg1,
                                    compare_int64_micros_arg1,
                                    compare_int64_micro_arg1,
                                    copy_int64_from_serialized,
                                    copy_int64_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_set_set(validate_set_int64, ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET_INTERSECT && num_ops == 1)
    {
        return apply_set_intersect(validate_set_int64, validate_int64,
                                   compare_int64, copy_int64_from_serialized,
                                   old_value, ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET_UNION && num_ops == 1)
    {
        return apply_set_union(validate_set_int64, validate_int64,
                               compare_int64, copy_int64_from_serialized,
                               old_value, ops, writeto, error);
    }
    else
    {
        std::cout<<"flag 43"<<std::endl;
        *error = hyperdex::NET_BADMICROS;
        return NULL;
    }
}

////////////////////////// Instantiated Map Functions //////////////////////////

static bool
validate_map_string_string(const e::slice& map)
{
    return validate_map(validate_string, validate_string, compare_string, map);
}

static bool
validate_map_string_int64(const e::slice& map)
{
    return validate_map(validate_string, validate_int64, compare_string, map);
}

static bool
validate_map_int64_string(const e::slice& map)
{
    return validate_map(validate_int64, validate_string, compare_int64, map);
}

static bool
validate_map_int64_int64(const e::slice& map)
{
    return validate_map(validate_int64, validate_int64, compare_int64, map);
}

// This wrapper is needed because "apply_string operates on string attributes,
// which do not encode the size before the string because every attribute has an
// implicit "size" argument.  However, when applying the string to something in
// a map, we need to also encode the size.
static uint8_t*
apply_string_wrapper(const e::slice& old_value,
                     const hyperdex::microop* ops,
                     size_t num_ops,
                     uint8_t* writeto,
                     hyperdex::network_returncode* error)
{
    uint8_t* original_writeto = writeto;
    writeto = apply_string(old_value, ops, num_ops, writeto + sizeof(uint32_t), error);
    e::pack32le(writeto - original_writeto - sizeof(uint32_t), original_writeto);
    return writeto;
}

static uint8_t*
apply_map_string_string(const e::slice& old_value,
                        hyperdex::microop* ops,
                        size_t num_ops,
                        uint8_t* writeto,
                        hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_MAP_ADD ||
        ops[0].action == hyperdex::OP_MAP_REMOVE)
    {
        return apply_map_add_remove(validate_string,
                                    validate_string,
                                    validate_string_micro,
                                    validate_string_micro,
                                    compare_string_micros_arg2,
                                    compare_string_micro_arg2,
                                    copy_string_from_serialized,
                                    copy_string_from_micro_arg2,
                                    copy_string_from_serialized,
                                    copy_string_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_map_set(validate_map_string_string, ops, writeto, error);
    }
    else
    {
        return apply_map_microop(validate_string,
                                 validate_string,
                                 validate_string_micro,
                                 validate_string_micro,
                                 compare_string_micros_arg2,
                                 compare_string_micro_arg2,
                                 copy_string_from_serialized,
                                 copy_string_from_micro_arg2,
                                 copy_string_from_serialized,
                                 string_slice_from_serialized,
                                 apply_string_wrapper,
                                 old_value, ops, num_ops, writeto, error);
    }
}

static uint8_t*
apply_map_string_int64(const e::slice& old_value,
                       hyperdex::microop* ops,
                       size_t num_ops,
                       uint8_t* writeto,
                       hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_MAP_ADD ||
        ops[0].action == hyperdex::OP_MAP_REMOVE)
    {
        return apply_map_add_remove(validate_string,
                                    validate_int64,
                                    validate_int64_micro_arg1,
                                    validate_string_micro,
                                    compare_string_micros_arg2,
                                    compare_string_micro_arg2,
                                    copy_string_from_serialized,
                                    copy_string_from_micro_arg2,
                                    copy_int64_from_serialized,
                                    copy_int64_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_map_set(validate_map_string_int64, ops, writeto, error);
    }
    else
    {
        return apply_map_microop(validate_string,
                                 validate_int64,
                                 validate_int64_micro_arg1,
                                 validate_string_micro,
                                 compare_string_micros_arg2,
                                 compare_string_micro_arg2,
                                 copy_string_from_serialized,
                                 copy_string_from_micro_arg2,
                                 copy_int64_from_serialized,
                                 int64_slice_from_serialized,
                                 apply_int64,
                                 old_value, ops, num_ops, writeto, error);
    }
}

static uint8_t*
apply_map_int64_string(const e::slice& old_value,
                       hyperdex::microop* ops,
                       size_t num_ops,
                       uint8_t* writeto,
                       hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_MAP_ADD ||
        ops[0].action == hyperdex::OP_MAP_REMOVE)
    {
        return apply_map_add_remove(validate_int64,
                                    validate_string,
                                    validate_string_micro,
                                    validate_int64_micro_arg2,
                                    compare_int64_micros_arg2,
                                    compare_int64_micro_arg2,
                                    copy_int64_from_serialized,
                                    copy_int64_from_micro_arg2,
                                    copy_string_from_serialized,
                                    copy_string_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_map_set(validate_map_int64_string, ops, writeto, error);
    }
    else
    {
        return apply_map_microop(validate_int64,
                                 validate_string,
                                 validate_string_micro,
                                 validate_int64_micro_arg2,
                                 compare_int64_micros_arg2,
                                 compare_int64_micro_arg2,
                                 copy_int64_from_serialized,
                                 copy_int64_from_micro_arg2,
                                 copy_string_from_serialized,
                                 string_slice_from_serialized,
                                 apply_string_wrapper,
                                 old_value, ops, num_ops, writeto, error);
    }
}

static uint8_t*
apply_map_int64_int64(const e::slice& old_value,
                      hyperdex::microop* ops,
                      size_t num_ops,
                      uint8_t* writeto,
                      hyperdex::network_returncode* error)
{
    assert(num_ops > 0);

    if (ops[0].action == hyperdex::OP_MAP_ADD ||
        ops[0].action == hyperdex::OP_MAP_REMOVE)
    {
        return apply_map_add_remove(validate_int64,
                                    validate_int64,
                                    validate_int64_micro_arg1,
                                    validate_int64_micro_arg2,
                                    compare_int64_micros_arg2,
                                    compare_int64_micro_arg2,
                                    copy_int64_from_serialized,
                                    copy_int64_from_micro_arg2,
                                    copy_int64_from_serialized,
                                    copy_int64_from_micro_arg1,
                                    old_value, ops, num_ops, writeto, error);
    }
    else if (ops[0].action == hyperdex::OP_SET && num_ops == 1)
    {
        return apply_map_set(validate_map_int64_int64, ops, writeto, error);
    }
    else
    {
        return apply_map_microop(validate_int64,
                                 validate_int64,
                                 validate_int64_micro_arg2,
                                 validate_int64_micro_arg1,
                                 compare_int64_micros_arg2,
                                 compare_int64_micro_arg2,
                                 copy_int64_from_serialized,
                                 copy_int64_from_micro_arg2,
                                 copy_int64_from_serialized,
                                 int64_slice_from_serialized,
                                 apply_int64,
                                 old_value, ops, num_ops, writeto, error);
    }
}

/////////////////////////////// Public Functions ///////////////////////////////

bool
hyperdaemon :: validate_datatype(hyperdatatype datatype, const e::slice& data)
{
    switch (datatype)
    {
        case HYPERDATATYPE_STRING:
            return true;
        case HYPERDATATYPE_INT64:
            return data.size() == 0 || data.size() == sizeof(int64_t);
        case HYPERDATATYPE_LIST_STRING:
            return validate_list_string(data);
        case HYPERDATATYPE_LIST_INT64:
            return validate_list_int64(data);
        case HYPERDATATYPE_SET_STRING:
            return validate_set_string(data);
        case HYPERDATATYPE_SET_INT64:
            return validate_set_int64(data);
        case HYPERDATATYPE_MAP_STRING_STRING:
            return validate_map_string_string(data);
        case HYPERDATATYPE_MAP_STRING_INT64:
            return validate_map_string_int64(data);
        case HYPERDATATYPE_MAP_INT64_STRING:
            return validate_map_int64_string(data);
        case HYPERDATATYPE_MAP_INT64_INT64:
            return validate_map_int64_int64(data);
        case HYPERDATATYPE_LIST_GENERIC:
        case HYPERDATATYPE_SET_GENERIC:
        case HYPERDATATYPE_MAP_GENERIC:
        case HYPERDATATYPE_MAP_STRING_KEYONLY:
        case HYPERDATATYPE_MAP_INT64_KEYONLY:
        case HYPERDATATYPE_GARBAGE:
            return false;
        default:
            abort();
    }
}

size_t
hyperdaemon :: sizeof_microop(const hyperdex::microop& op)
{
    return 2 * (sizeof(int64_t) + sizeof(uint32_t))
           + op.arg1.size() + op.arg2.size();
}

uint8_t*
hyperdaemon :: apply_microops(hyperdatatype type,
                              const e::slice& old_value,
                              hyperdex::microop* ops,
                              size_t num_ops,
                              uint8_t* writeto,
                              hyperdex::network_returncode* error)
{
    switch (type)
    {
        case HYPERDATATYPE_STRING:
            return apply_string(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_INT64:
            return apply_int64(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_LIST_STRING:
            return apply_list_string(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_LIST_INT64:
            return apply_list_int64(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_SET_STRING:
            return apply_set_string(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_SET_INT64:
            return apply_set_int64(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_MAP_STRING_STRING:
            return apply_map_string_string(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_MAP_STRING_INT64:
            return apply_map_string_int64(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_MAP_INT64_STRING:
            return apply_map_int64_string(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_MAP_INT64_INT64:
            return apply_map_int64_int64(old_value, ops, num_ops, writeto, error);
        case HYPERDATATYPE_LIST_GENERIC:
        case HYPERDATATYPE_SET_GENERIC:
        case HYPERDATATYPE_MAP_GENERIC:
        case HYPERDATATYPE_MAP_STRING_KEYONLY:
        case HYPERDATATYPE_MAP_INT64_KEYONLY:
        case HYPERDATATYPE_GARBAGE:
        default:
            std::cout<<"flag 44"<<std::endl;
            *error = hyperdex::NET_BADMICROS;
            return NULL;
    }
}

redpajama_set_name: RedPajamaGithub


text:
Liana Pailodze is a versatile pianist who performs both contemporary and classical styles of music.
She has performed as a solo artist since the age of 8, and has participated in chamber music settings from the age of 11. As a soloist, she has been featured with the Shilakadze Chamber Orchestra, the Georgian Television Orchestra, the Georgian Symphonic Orchestra, and in 2007 she performed Stravinsky's Concerto for Piano and Wind Instruments with Frost Wind Ensemble. In addition to her solo activities she has performed in different chamber settings in venues that include Carnegie Hall, Benaroya Hall, Arsht Center, Lincoln Theatre, the New World Center.
Born in Tbilisi, Republic of Georgia, Liana first studied piano at the Paliashvili Special Music School for Gifted Children. She had a wide range of interests at a young age, resulting in a Diploma in Chemical Analysis at the Tbilisi Aviation Machine Building School in 1998. She grew up in a unique musical setting. Her father is a prominent jazz drummer, and her mother, a computer engineer, was her first piano teacher. The strong presence of music in Georgian culture continues to have a strong effect on her musical identity.
Currently acting vice president of the Miami Piano Circle, she is also adjunct faculty at Barry University as well as a doctoral candidate at the University of Miami's Frost School of Music.
http://www.lianapailodze.com/lianapailodze/Home.html
Liana Pailodze, piano and percussion
Dan Dickinson, Bass
Dan Dickinson is a composer and bassist living in Miami, Florida.
Dickinson is active in many creative expressions, including interactive soundscape installations, experimental improvisation, and chamber music. He holds a DMA in composition from the University of Miami, and is a founding member of the improvisational ensemble Fridamusiq.
http://www.zuletaarts.com
© 2014 by MATTHEW EVAN TAYLOR.
MATTHEW EVAN TAYLOR